Lab 02 — Ingestion — Land Raw to Bronze

The goal

In Lab 01 your sources came alive. In this lab you build ingestion — the ingest/extract_postgres.py script that copies source tables out of Postgres and lands them into a bronze schema inside a single DuckDB file (warehouse.duckdb). That file becomes your warehouse for the rest of the capstone: silver and gold (Lab 03) build on top of bronze.

Two properties matter more than anything else here, and they're what separate a toy script from a pipeline:

• Idempotent — running it twice leaves the warehouse in the same state as running it once. No duplicate rows, ever.
• Incremental — the big, growing table (rentals) is pulled using a high-watermark, so a second run only fetches rows newer than what you already have. Cheap to re-run.

What "bronze" means

The medallion architecture (bronze → silver → gold) organizes a warehouse into layers of increasing refinement. Bronze is the landing zone: a near-exact copy of the source, captured as-is, with as little transformation as possible. You don't clean it, rename columns, or fix types here — you just get it in, reliably. The one thing we add is a loaded_at stamp on every row: when it landed. Lab 06 reads that column to tell whether a feed has gone quiet.

Let's create the warehouse file and the bronze schema. DuckDB is a single file — creating it is just connecting to a path that doesn't exist yet.

# from the repo root, create the schema in a fresh warehouse file
duckdb warehouse.duckdb "CREATE SCHEMA IF NOT EXISTS bronze;"

# confirm it exists
duckdb warehouse.duckdb "SELECT schema_name FROM information_schema.schemata;"

Writing the extractor

Here's the plan. The three small reference tables (customers, gpus, gpu_prices) change slowly and are cheap, so we full-refresh them: drop and replace on every run. That's trivially idempotent. The rentals table grows continuously, so we pull it incrementally using a high-watermark — we remember the largest rental_id we've already landed, and ask Postgres only for rows above it.

We'll use DuckDB's Postgres scanner extension, which lets DuckDB read Postgres tables directly inside a SQL query — no separate Postgres driver, no row-by-row copying in Python. Create the file:

"""Extract source tables from Postgres and land them in the bronze layer.

Full-refresh for small reference tables; high-watermark incremental for rentals.
Safe to re-run: idempotent by construction.
"""
import os
import duckdb

WAREHOUSE = os.getenv("WAREHOUSE", "warehouse.duckdb")

# Connection string for DuckDB's postgres scanner. Matches the Lab 01 Compose file.
PG = os.getenv(
    "PG_CONN",
    "host=localhost port=5432 dbname=marketplace user=griddp password=griddp",
)

# Small tables we simply replace every run -> trivially idempotent.
FULL_REFRESH = ["customers", "gpus", "gpu_prices"]


def connect():
    con = duckdb.connect(WAREHOUSE)
    con.execute("INSTALL postgres; LOAD postgres;")
    # Attach Postgres as a read-only catalog called `pg`.
    con.execute(f"ATTACH '{PG}' AS pg (TYPE postgres, READ_ONLY);")
    con.execute("CREATE SCHEMA IF NOT EXISTS bronze;")
    return con


def full_refresh(con, table):
    """Replace a bronze table with a fresh copy from Postgres, stamped with the load time."""
    con.execute(f"CREATE OR REPLACE TABLE bronze.{table} AS "
                f"SELECT *, now() AS loaded_at FROM pg.public.{table};")
    n = con.execute(f"SELECT count(*) FROM bronze.{table}").fetchone()[0]
    print(f"  full-refresh bronze.{table:12s} -> {n} rows")


def ensure_watermark_store(con):
    """A tiny table that remembers how far we've ingested each source."""
    con.execute("""
        CREATE TABLE IF NOT EXISTS bronze._watermarks (
            source_table VARCHAR PRIMARY KEY,
            high_watermark BIGINT
        );
    """)


def get_watermark(con, table):
    row = con.execute(
        "SELECT high_watermark FROM bronze._watermarks WHERE source_table = ?",
        [table],
    ).fetchone()
    return row[0] if row else 0


def set_watermark(con, table, value):
    con.execute("""
        INSERT INTO bronze._watermarks VALUES (?, ?)
        ON CONFLICT (source_table) DO UPDATE SET high_watermark = excluded.high_watermark;
    """, [table, value])


def incremental_rentals(con):
    """Pull only rentals newer than the last high-watermark (max rental_id)."""
    con.execute("""
        CREATE TABLE IF NOT EXISTS bronze.rentals AS
        SELECT *, now() AS loaded_at FROM pg.public.rentals WHERE false;
    """)
    wm = get_watermark(con, "rentals")

    new_rows = con.execute(
        "SELECT count(*) FROM pg.public.rentals WHERE rental_id > ?", [wm]
    ).fetchone()[0]

    con.execute(
        "INSERT INTO bronze.rentals "
        "SELECT *, now() FROM pg.public.rentals WHERE rental_id > ?", [wm]
    )

    new_wm = con.execute(
        "SELECT max(rental_id) FROM bronze.rentals"
    ).fetchone()[0] or 0
    set_watermark(con, "rentals", new_wm)

    total = con.execute("SELECT count(*) FROM bronze.rentals").fetchone()[0]
    print(f"  incremental bronze.rentals      -> +{new_rows} new "
          f"(watermark rental_id={new_wm}, total {total})")


def main():
    con = connect()
    ensure_watermark_store(con)
    print(f"Landing bronze into {WAREHOUSE} ...")
    for t in FULL_REFRESH:
        full_refresh(con, t)
    incremental_rentals(con)
    con.close()
    print("Done.")


if __name__ == "__main__":
    main()

Run it

Run the extractor from the repo root:

uv run python ingest/extract_postgres.py

Landing bronze into warehouse.duckdb ...
  full-refresh bronze.customers    -> 50 rows
  full-refresh bronze.gpus         -> 10 rows
  full-refresh bronze.gpu_prices   -> 23 rows
  incremental bronze.rentals      -> +400 new (watermark rental_id=400, total 400)
Done.

Run it twice (idempotency)

This is the test that proves your pipeline is safe. Run the exact same command a second time, without generating any new data:

uv run python ingest/extract_postgres.py

  full-refresh bronze.customers    -> 50 rows
  full-refresh bronze.gpus         -> 10 rows
  full-refresh bronze.gpu_prices   -> 23 rows
  incremental bronze.rentals      -> +0 new (watermark rental_id=400, total 400)
Done.

docker compose exec postgres psql -U griddp -d marketplace -c "
  INSERT INTO rentals (customer_id, gpu_id, hours, started_at)
  SELECT customer_id, gpu_id, 2.0, now() FROM rentals ORDER BY rental_id LIMIT 5;"
uv run python ingest/extract_postgres.py

Verify the bronze layer

Open the warehouse and look at what landed. DuckDB's CLI gives you a SQL prompt straight against the file:

duckdb warehouse.duckdb

-- what tables landed in bronze?
SELECT table_name FROM information_schema.tables
WHERE table_schema = 'bronze' ORDER BY table_name;

-- row counts per bronze table
SELECT 'customers'  AS t, count(*) FROM bronze.customers
UNION ALL SELECT 'gpus',       count(*) FROM bronze.gpus
UNION ALL SELECT 'gpu_prices', count(*) FROM bronze.gpu_prices
UNION ALL SELECT 'rentals',    count(*) FROM bronze.rentals;

-- the watermark we're tracking
SELECT * FROM bronze._watermarks;

-- peek at a few landed rentals
SELECT rental_id, customer_id, gpu_id, started_at
FROM bronze.rentals ORDER BY rental_id DESC LIMIT 5;

Commit Lab 02

You've added the ingestion stage. Commit the code (the warehouse file stays gitignored):

git add ingest/extract_postgres.py .gitignore
git status                       # confirm warehouse.duckdb is NOT staged
git commit -m "Lab 02: ingest Postgres -> bronze (incremental + idempotent)"

✓ Check yourself

Answer from memory before opening each one.

1. A teammate wants the extractor to trim whitespace and drop obviously broken rows on the way into bronze, "to save dbt the work". What is lost if you agree?

   Show answer

   The record of what the source actually said. If that cleaning rule turns out to be wrong, the discarded detail is gone and you have to re-pull from the source; with a raw bronze you fix the transform and rebuild silver and gold from what already landed. Cleaning belongs in silver. See What "bronze" means.

2. Run the extractor twice with no new source data. What should the rentals line print the second time — and if the total had doubled instead, where would you look?

   Show answer

   +0 new, with the same watermark and the same total: the stored watermark already equals the largest rental_id, so rental_id > watermark matches nothing. A doubled total means the insert is not being filtered — the watermark was never stored or advanced (check bronze._watermarks), or the INSERT lost its WHERE. See Run it twice.

3. Someone switches the rentals watermark from rental_id to started_at, reasoning that a timestamp is more meaningful. Which rentals does the extractor start missing?

   Show answer

   Any rental inserted later but carrying a started_at earlier than the stored maximum. A watermark needs a column that only ever increases as rows are added. rental_id does; started_at does not — the Lab 01 generator writes rentals scattered across the last 90 days in random order — so new rows routinely fall below the watermark and are skipped for good. See Writing the extractor.

4. A customer is upgraded from free to pro in Postgres. After the next extractor run, what does bronze.customers show — and what would it show had customers used the same watermark pattern as rentals?

   Show answer

   It shows pro: CREATE OR REPLACE TABLE re-copies the whole table each run, so updates come along for free (the old value is overwritten, not kept). A customer_id watermark only fetches ids above the stored maximum, so an edit to an existing row would never be picked up and bronze would say free indefinitely. See Writing the extractor.

5. The process is killed after INSERT INTO bronze.rentals finishes but before set_watermark runs. What does the next run do, and how would you close the gap?

   Show answer

   It reads the old watermark from _watermarks, selects the same rows again and inserts them a second time — duplicates. The insert and the watermark update are separate statements with no transaction around them, so "run it twice" only proves idempotency for runs that finish. Wrap both in one transaction, or take the watermark from max(rental_id) in bronze.rentals at the start of each run instead of from a separately stored value. See incremental_rentals.

6. customers grows to 50 million rows and the nightly full refresh now takes an hour. What do you weigh before moving it to the watermark pattern used for rentals?

   Show answer

   Size now argues for incremental, but the append-only watermark is only correct for rows that never change — and customers do (a tier upgrade). A plain id watermark would be fast and wrong; you need a way to catch edits too, which means a merge/upsert on the primary key driven by a change timestamp. Full refresh stays the right call for as long as re-copying is cheap, because it is trivially idempotent and picks up updates. See Writing the extractor.

def naive_rentals(con):
    con.execute("CREATE OR REPLACE TABLE bronze.rentals AS "
                "SELECT * FROM pg.public.rentals;")

def incremental_rentals(con):
    # 1) make sure the target exists (empty, same shape as source)
    con.execute("CREATE TABLE IF NOT EXISTS bronze.rentals AS "
                "SELECT * FROM pg.public.rentals WHERE false;")

    # 2) read how far we've already ingested (0 the first time)
    wm = get_watermark(con, "rentals")

    # 3) insert ONLY rows newer than the watermark
    con.execute("INSERT INTO bronze.rentals "
                "SELECT * FROM pg.public.rentals WHERE rental_id > ?", [wm])

    # 4) advance the watermark to the new max
    new_wm = con.execute(
        "SELECT max(rental_id) FROM bronze.rentals").fetchone()[0] or 0
    set_watermark(con, "rentals", new_wm)

Takeaway

• Bronze is a near-exact copy of the source plus a load stamp, kept so any downstream bug can be fixed by replaying transforms instead of re-pulling.
• Small reference tables are fully refreshed with CREATE OR REPLACE, which is idempotent by construction, and the growing rentals table is pulled incrementally past a stored high watermark.
• Idempotency is something you can see: run the extractor twice and the counts stay put (400 rows, not 800) with zero new rentals reported.
• Insert five rentals straight into Postgres and the next run reports five new rows, which proves the incremental path moves only what changed.
• Do not use the generator to add rows after ingestion, because it drops and recreates every table, the ids restart at 1, and the watermark then skips them.
• The warehouse file is generated data, so it goes in .gitignore and you commit the code that builds it.

Bronze is a faithful but messy copy, and the next lab cleans it, types it and shapes it into a star schema. → Lab 03 — Transform: Silver & Gold with dbt

Further reading

• DuckDB: PostgreSQL extension — DuckDB docs, accessed 2026. Attaching a Postgres database and reading its tables from SQL.
• DuckDB: CREATE TABLE — DuckDB docs, accessed 2026. CREATE OR REPLACE TABLE and constraints.
• What is a medallion architecture? — Databricks. The vendor definition of the bronze layer's role.
• Designing Data-Intensive Applications — Kleppmann, O'Reilly. The background on why re-runnable, replayable loads matter.