CDC Lakehouse Reliability — E-commerce

submission/ecommerce-cdc/.gitignore

@@ -0,0 +1,6 @@
+__pycache__/
+*.pyc
+.pytest_cache/
+.venv/
+*.duckdb
+.data/

submission/ecommerce-cdc/PR_DESCRIPTION.md

@@ -0,0 +1,70 @@
+# CDC Lakehouse Reliability — E-commerce
+
+A small but deeply correct CDC pipeline keeping a lake (full history) and a warehouse
+(latest snapshot + time travel) in sync with an e-commerce source.
+Stack: PostgreSQL (source design) · Python (ingestion) · DuckDB (lake + warehouse) ·
+pytest. CDC is simulated in-process (dependency-free); the production analogue is
+documented throughout. Full design: `submission/ecommerce-cdc/SOLUTION_APPROACH.md`.
+
+## 1. Source schema design
+- **Tables:** `customers`, `products`, `orders` (strong); `order_items`, `payments` (weak).
+- **Strong vs weak:** strong entities have independent identity; `order_items` and
+  `payments` have no meaning without their parent `order` (identity completed by the FK).
+- **Keys/relationships:** PKs on every table; FKs `orders→customers`,
+  `order_items→orders/products`, `payments→orders` (1—N throughout).
+- **Indexes:** FK columns + `updated_at` (the change-capture path).
+- **Types:** decimals (`price/total/amount/line_total`), timestamps, enums
+  (`status/method`), nullable (`settled_at`).
+- **Validation rules:** `total = Σ line_total`, `line_total = qty*unit_price`,
+  non-negative money, `price>0`, `quantity>0`, `settled_at ≥ created_at`, enum domains,
+  legal status transitions.
+
+## 2. CDC strategy
+- Every mutation → a `CDCRecord(op, table, pk, before, after, captured_at, sequence)`.
+- **`sequence`** is a monotonic offset (Kafka offset / Postgres LSN analogue) — the sole
+  ordering + checkpoint primitive.
+- **Replay/restart:** checkpoint stores the last applied sequence; restart replays only
+  `records_since(checkpoint)`.
+- **Duplicates:** lake PK + `ON CONFLICT DO NOTHING`; warehouse skips events whose
+  sequence ≤ the row's current `_cdc_seq` (also handles out-of-order).
+- **Deletes:** captured as `delete` events; snapshot soft-deletes, SCD2 history closes
+  the version.
+
+## 3. Lake & warehouse modeling
+- **Lake (`lake_cdc_events`):** append-only, idempotent, stores full row image as JSON —
+  every change is recoverable.
+- **Warehouse current (`wh_*`):** upsert by PK + `_deleted` soft-delete = latest snapshot.
+- **Warehouse history (`wh_*_history`):** SCD2 (`_valid_from_seq`/`_valid_to_seq`/
+  `_is_current`).
+- **Time travel:** `reconstruct_as_of(table, seq)` reads the version valid at `seq`.
+- **Restore:** `restore_from_lake(target_seq)` deterministically rebuilds the snapshot by
+  replaying the lake.
+
+## 4. Schema change safety
+- Declared `SCHEMA_CONTRACT` (columns, types, nullability, enum domains, keys).
+- Each cycle diffs live source schema vs contract. **Breaking** (dropped/renamed column,
+  type change, nullable tightening, enum shrink) → raise `SchemaIncompatibilityError`,
+  **stop ingestion, write nothing**, exit non-zero. **Additive** changes warn and continue.
+- CI gate: `scripts/check_schema_contracts.py`.
+
+## 5. Validation parity
+- **System:** PK uniqueness, not-null, referential integrity, enum domains.
+- **Business:** total = Σ line_total, line_total = qty*unit_price, non-negative money,
+  `price>0`, `quantity>0`, `settled_at ≥ created_at`, legal status transitions (validated
+  from ordered lake history).
+- Surfaced via `validation/checks.py`, `scripts/run_data_quality_checks.py`, and tests.
+
+## 6. Catalog exposure
+- `catalog/catalog.json` registers every lake + warehouse dataset (layer, owner,
+  consumers, cadence, query path, per-column schema). Gate: `scripts/validate_catalog.py`;
+  a test also asserts catalog ↔ live tables stay in sync.
+
+## 7. Responsible AI usage
+AI assistance was used to scaffold boilerplate and draft documentation. All schema
+design, CDC/SCD2 semantics, schema-safety rules, and validation logic were reviewed and
+verified by running the test suite (`41 passed`) and the CI gate scripts locally.
+
+## Testing
+`pytest -q` → **41 passed**, covering modeling/constraints, CDC correctness
+(insert/update/delete/duplicate/replay/restart), schema-change safety, warehouse
+snapshot + time travel + restore, and catalog. Red→Blue→Green discipline followed.

submission/ecommerce-cdc/README.md

@@ -0,0 +1,88 @@
+# E-commerce CDC Lakehouse
+
+A change-data-capture pipeline that keeps a **lake** (full change history) and a
+**warehouse** (latest snapshot + time travel) in sync with an e-commerce source
+system, with schema-drift safety and validation parity.
+
+> Design rationale and the full step-by-step approach live in
+> [SOLUTION_APPROACH.md](./SOLUTION_APPROACH.md).
+
+## Stack
+
+- **PostgreSQL** — canonical source design (`source/postgres_ddl.sql`)
+- **Python** — CDC capture, checkpointing, schema safety, orchestration
+- **DuckDB** — lake + warehouse (object-storage / cloud-warehouse analogue)
+- **pytest** — system + business validation
+
+CDC runs in two interchangeable modes:
+
+- **Simulated in-process** (default) — runs with no external services; used by the test
+  suite and CI.
+- **Real PostgreSQL source via Docker** (optional) — insert/update/delete in a real
+  Postgres and capture the changes into the lake + warehouse. See
+  [TESTING_POSTGRES.md](./TESTING_POSTGRES.md).
+
+The production analogue (Debezium / Postgres WAL) is documented at every layer.
+
+## Layout
+
+```
+source/      Postgres DDL + DuckDB models + SCHEMA_CONTRACT
+ingestion/   cdc capture · checkpoint · schema-contract safe-stop · pipeline · seed
+lake/        append-only, idempotent CDC event history
+warehouse/   current snapshot + SCD2 history + restore / time travel
+validation/  system + business parity checks
+catalog/     catalog.json dataset registry
+postgres/    optional real-Postgres source bridge (snapshot-diff CDC via psycopg)
+scripts/     CI gates (schema · data quality · catalog) + Postgres demo CLIs
+tests/       red/blue/green tests across all categories
+docker-compose.yml   local Postgres that auto-loads the source schema
+```
+
+## Setup & run (simulated, no external services)
+
+```bash
+cd submission/ecommerce-cdc
+python -m venv .venv && . .venv/bin/activate
+pip install -r requirements.txt
+```
+
+```bash
+pytest -q                                  # full test suite (41 tests)
+python -m ingestion.pipeline               # end-to-end demo (ingest + time travel + restore)
+
+python scripts/check_schema_contracts.py   # schema safety gate
+python scripts/run_data_quality_checks.py  # validation parity gate
+python scripts/validate_catalog.py         # catalog gate
+```
+
+## Run against a real PostgreSQL source (optional, Docker)
+
+Requires Docker (e.g. Colima); the Postgres driver (`psycopg`) is already in
+`requirements.txt`. Full walkthrough: [TESTING_POSTGRES.md](./TESTING_POSTGRES.md).
+
+```bash
+docker-compose up -d                        # start Postgres (schema auto-created)
+
+python scripts/pg_seed.py                   # insert realistic sample rows into Postgres
+python scripts/pg_sync.py                   # capture changes -> lake + warehouse + report
+python scripts/pg_mutate.py                 # apply updates/deletes (legal transitions)
+python scripts/pg_sync.py                   # re-sync to see update/delete capture
+```
+
+Postgres connection: `localhost:5432`, db `kulu_ecom`, user/pass `kulu`/`kulu`.
+The persistent lakehouse lives at `.data/lakehouse.duckdb`.
+
+## What it demonstrates
+
+| Requirement | Where |
+|---|---|
+| Source model (strong/weak entities, keys, indexes, enums) | `source/` |
+| CDC capture + checkpoint/replay/restart | `ingestion/cdc.py`, `ingestion/checkpoint.py` |
+| Lake retains every change (idempotent) | `lake/lake.py` |
+| Warehouse latest snapshot + SCD2 time travel + restore | `warehouse/warehouse.py` |
+| Schema-change detection + fail-closed stop | `ingestion/schema_contract.py` |
+| Validation parity (system + business) | `validation/checks.py` |
+| Catalog exposure | `catalog/catalog.json` |
+| Reliability (duplicates, deletes, out-of-order, restart) | see SOLUTION_APPROACH §10 |
+| Real Postgres source + CDC capture (optional) | `postgres/`, `scripts/pg_*.py`, `docker-compose.yml` |