Lab 02 — Ingestion — Land Raw to Bronze

# 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;"

"""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 your Lab 01 .env.
PG = os.getenv(
    "PG_CONN",
    "host=localhost port=5432 dbname=griddp 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."""
    con.execute(f"CREATE OR REPLACE TABLE bronze.{table} AS "
                f"SELECT * 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 * 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 * 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()

uv run python ingest/extract_postgres.py

uv run python ingest/extract_postgres.py

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;

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)"