diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..f40f450 --- /dev/null +++ b/.gitignore @@ -0,0 +1,4 @@ +.DS_Store +__pycache__/ +.pytest_cache/ +*.pyc diff --git a/lake/customer.jsonl b/lake/customer.jsonl new file mode 100644 index 0000000..775a406 --- /dev/null +++ b/lake/customer.jsonl @@ -0,0 +1,4 @@ +{"table": "customer", "op": "upsert", "pk": 1, "updated_at": "2026-08-15 19:01:35.384343+00:00", "captured_at": "2026-08-16T05:25:12.835026+00:00", "data": {"id": 1, "name": "Asha Rao", "email": "asha.rao@example.com", "created_at": "2026-08-15 19:01:35.384343+00:00", "updated_at": "2026-08-15 19:01:35.384343+00:00", "is_deleted": false}} +{"table": "customer", "op": "upsert", "pk": 2, "updated_at": "2026-08-15 19:01:35.384343+00:00", "captured_at": "2026-08-16T05:25:12.836577+00:00", "data": {"id": 2, "name": "Vikram Shah", "email": "vikram.shah@example.com", "created_at": "2026-08-15 19:01:35.384343+00:00", "updated_at": "2026-08-15 19:01:35.384343+00:00", "is_deleted": false}} +{"table": "customer", "op": "upsert", "pk": 3, "updated_at": "2026-08-15 19:01:35.384343+00:00", "captured_at": "2026-08-16T05:25:12.836619+00:00", "data": {"id": 3, "name": "Priya Nair", "email": "priya.nair@example.com", "created_at": "2026-08-15 19:01:35.384343+00:00", "updated_at": "2026-08-15 19:01:35.384343+00:00", "is_deleted": false}} +{"table": "customer", "op": "upsert", "pk": 4, "updated_at": "2026-08-16 05:28:15.526971+00:00", "captured_at": "2026-08-16T05:28:59.517745+00:00", "data": {"id": 4, "name": "Rohit Verma", "email": "rohit.verma@example.com", "created_at": "2026-08-16 05:28:15.526971+00:00", "updated_at": "2026-08-16 05:28:15.526971+00:00", "is_deleted": false}} diff --git a/lake/payment_attempt.jsonl b/lake/payment_attempt.jsonl new file mode 100644 index 0000000..781d839 --- /dev/null +++ b/lake/payment_attempt.jsonl @@ -0,0 +1,4 @@ +{"table": "payment_attempt", "op": "upsert", "pk": 1, "updated_at": "2026-08-15 19:01:35.408173+00:00", "captured_at": "2026-08-16T05:25:12.873971+00:00", "data": {"id": 1, "transfer_id": 1, "attempt_no": 1, "status": "succeeded", "failure_reason": null, "created_at": "2026-08-15 19:01:35.408173+00:00", "updated_at": "2026-08-15 19:01:35.408173+00:00", "is_deleted": false}} +{"table": "payment_attempt", "op": "upsert", "pk": 2, "updated_at": "2026-08-15 19:01:35.409717+00:00", "captured_at": "2026-08-16T05:25:12.874017+00:00", "data": {"id": 2, "transfer_id": 2, "attempt_no": 1, "status": "initiated", "failure_reason": null, "created_at": "2026-08-15 19:01:35.409717+00:00", "updated_at": "2026-08-15 19:01:35.409717+00:00", "is_deleted": false}} +{"table": "payment_attempt", "op": "upsert", "pk": 2, "updated_at": "2026-08-16 05:28:15.519337+00:00", "captured_at": "2026-08-16T05:28:59.525802+00:00", "data": {"id": 2, "transfer_id": 2, "attempt_no": 1, "status": "succeeded", "failure_reason": null, "created_at": "2026-08-15 19:01:35.409717+00:00", "updated_at": "2026-08-16 05:28:15.519337+00:00", "is_deleted": false}} +{"table": "payment_attempt", "op": "upsert", "pk": 3, "updated_at": "2026-08-16 05:28:15.528979+00:00", "captured_at": "2026-08-16T05:28:59.525810+00:00", "data": {"id": 3, "transfer_id": 3, "attempt_no": 1, "status": "failed", "failure_reason": "insufficient_funds", "created_at": "2026-08-16 05:28:15.528979+00:00", "updated_at": "2026-08-16 05:28:15.528979+00:00", "is_deleted": false}} diff --git a/lake/transfer.jsonl b/lake/transfer.jsonl new file mode 100644 index 0000000..1993fa2 --- /dev/null +++ b/lake/transfer.jsonl @@ -0,0 +1,4 @@ +{"table": "transfer", "op": "upsert", "pk": 1, "updated_at": "2026-08-15 19:01:35.405804+00:00", "captured_at": "2026-08-16T05:25:12.865067+00:00", "data": {"id": 1, "source_wallet_id": 1, "dest_wallet_id": 2, "amount": "500.00", "status": "settled", "created_at": "2026-08-15 19:01:35.405804+00:00", "settled_at": "2026-08-15 19:01:35.405804+00:00", "updated_at": "2026-08-15 19:01:35.405804+00:00", "is_deleted": false}} +{"table": "transfer", "op": "upsert", "pk": 2, "updated_at": "2026-08-15 19:01:35.409441+00:00", "captured_at": "2026-08-16T05:25:12.865114+00:00", "data": {"id": 2, "source_wallet_id": 3, "dest_wallet_id": 1, "amount": "1200.00", "status": "pending", "created_at": "2026-08-15 19:01:35.409441+00:00", "settled_at": null, "updated_at": "2026-08-15 19:01:35.409441+00:00", "is_deleted": false}} +{"table": "transfer", "op": "upsert", "pk": 2, "updated_at": "2026-08-16 05:28:15.511463+00:00", "captured_at": "2026-08-16T05:28:59.522437+00:00", "data": {"id": 2, "source_wallet_id": 3, "dest_wallet_id": 1, "amount": "1200.00", "status": "settled", "created_at": "2026-08-15 19:01:35.409441+00:00", "settled_at": "2026-08-16 05:28:15.511463+00:00", "updated_at": "2026-08-16 05:28:15.511463+00:00", "is_deleted": false}} +{"table": "transfer", "op": "upsert", "pk": 3, "updated_at": "2026-08-16 05:28:15.528268+00:00", "captured_at": "2026-08-16T05:28:59.522448+00:00", "data": {"id": 3, "source_wallet_id": 2, "dest_wallet_id": 4, "amount": "250.00", "status": "failed", "created_at": "2026-08-16 05:28:15.528268+00:00", "settled_at": null, "updated_at": "2026-08-16 05:28:15.528268+00:00", "is_deleted": false}} diff --git a/lake/wallet.jsonl b/lake/wallet.jsonl new file mode 100644 index 0000000..4ddfe97 --- /dev/null +++ b/lake/wallet.jsonl @@ -0,0 +1,4 @@ +{"table": "wallet", "op": "upsert", "pk": 1, "updated_at": "2026-08-15 19:01:35.396879+00:00", "captured_at": "2026-08-16T05:25:12.858577+00:00", "data": {"id": 1, "customer_id": 1, "currency": "INR", "status": "active", "created_at": "2026-08-15 19:01:35.396879+00:00", "updated_at": "2026-08-15 19:01:35.396879+00:00", "is_deleted": false}} +{"table": "wallet", "op": "upsert", "pk": 2, "updated_at": "2026-08-15 19:01:35.396879+00:00", "captured_at": "2026-08-16T05:25:12.858626+00:00", "data": {"id": 2, "customer_id": 2, "currency": "INR", "status": "active", "created_at": "2026-08-15 19:01:35.396879+00:00", "updated_at": "2026-08-15 19:01:35.396879+00:00", "is_deleted": false}} +{"table": "wallet", "op": "upsert", "pk": 3, "updated_at": "2026-08-15 19:01:35.396879+00:00", "captured_at": "2026-08-16T05:25:12.858663+00:00", "data": {"id": 3, "customer_id": 3, "currency": "INR", "status": "active", "created_at": "2026-08-15 19:01:35.396879+00:00", "updated_at": "2026-08-15 19:01:35.396879+00:00", "is_deleted": false}} +{"table": "wallet", "op": "delete", "pk": 4, "updated_at": "2026-08-16 05:28:15.530112+00:00", "captured_at": "2026-08-16T05:28:59.520491+00:00", "data": {"id": 4, "customer_id": 4, "currency": "INR", "status": "closed", "created_at": "2026-08-16 05:28:15.527609+00:00", "updated_at": "2026-08-16 05:28:15.530112+00:00", "is_deleted": true}} diff --git a/lake/wallet_balance_history.jsonl b/lake/wallet_balance_history.jsonl new file mode 100644 index 0000000..1765be9 --- /dev/null +++ b/lake/wallet_balance_history.jsonl @@ -0,0 +1,7 @@ +{"table": "wallet_balance_history", "op": "upsert", "pk": 1, "updated_at": "2026-08-15 19:01:35.402594+00:00", "captured_at": "2026-08-16T05:25:12.869734+00:00", "data": {"id": 1, "wallet_id": 1, "balance_after": "5000.00", "recorded_at": "2026-08-15 19:01:35.402594+00:00", "updated_at": "2026-08-15 19:01:35.402594+00:00", "is_deleted": false}} +{"table": "wallet_balance_history", "op": "upsert", "pk": 2, "updated_at": "2026-08-15 19:01:35.402594+00:00", "captured_at": "2026-08-16T05:25:12.869775+00:00", "data": {"id": 2, "wallet_id": 2, "balance_after": "3000.00", "recorded_at": "2026-08-15 19:01:35.402594+00:00", "updated_at": "2026-08-15 19:01:35.402594+00:00", "is_deleted": false}} +{"table": "wallet_balance_history", "op": "upsert", "pk": 3, "updated_at": "2026-08-15 19:01:35.402594+00:00", "captured_at": "2026-08-16T05:25:12.869810+00:00", "data": {"id": 3, "wallet_id": 3, "balance_after": "10000.00", "recorded_at": "2026-08-15 19:01:35.402594+00:00", "updated_at": "2026-08-15 19:01:35.402594+00:00", "is_deleted": false}} +{"table": "wallet_balance_history", "op": "upsert", "pk": 4, "updated_at": "2026-08-15 19:01:35.409132+00:00", "captured_at": "2026-08-16T05:25:12.869853+00:00", "data": {"id": 4, "wallet_id": 1, "balance_after": "4500.00", "recorded_at": "2026-08-15 19:01:35.409132+00:00", "updated_at": "2026-08-15 19:01:35.409132+00:00", "is_deleted": false}} +{"table": "wallet_balance_history", "op": "upsert", "pk": 5, "updated_at": "2026-08-15 19:01:35.409132+00:00", "captured_at": "2026-08-16T05:25:12.869890+00:00", "data": {"id": 5, "wallet_id": 2, "balance_after": "3500.00", "recorded_at": "2026-08-15 19:01:35.409132+00:00", "updated_at": "2026-08-15 19:01:35.409132+00:00", "is_deleted": false}} +{"table": "wallet_balance_history", "op": "upsert", "pk": 6, "updated_at": "2026-08-16 05:28:15.521377+00:00", "captured_at": "2026-08-16T05:28:59.524062+00:00", "data": {"id": 6, "wallet_id": 3, "balance_after": "8800.00", "recorded_at": "2026-08-16 05:28:15.521377+00:00", "updated_at": "2026-08-16 05:28:15.521377+00:00", "is_deleted": false}} +{"table": "wallet_balance_history", "op": "upsert", "pk": 7, "updated_at": "2026-08-16 05:28:15.521377+00:00", "captured_at": "2026-08-16T05:28:59.524071+00:00", "data": {"id": 7, "wallet_id": 1, "balance_after": "5700.00", "recorded_at": "2026-08-16 05:28:15.521377+00:00", "updated_at": "2026-08-16 05:28:15.521377+00:00", "is_deleted": false}} diff --git a/submission/megha/APPROACH.md b/submission/megha/APPROACH.md new file mode 100644 index 0000000..367307d --- /dev/null +++ b/submission/megha/APPROACH.md @@ -0,0 +1,115 @@ +# Approach — CDC Lakehouse Reliability Assignment + +This is an initial outline PR to confirm my understanding of the problem before I build the +full solution. It covers the domain I'll model, the planned architecture, and the key design +decisions I intend to make. Full implementation will follow in subsequent commits on this +same branch/PR. + +## 1. Understanding of the Problem + +The core ask is not "build an ETL pipeline" — it's "design a small but *safe* data platform" +that: + +- Captures every change (insert/update/delete) from a transactional source. +- Writes that change history durably to a **lake** (append-only, full history, replayable). +- Maintains a **warehouse** with the latest snapshot, updated near real-time. +- Detects **incompatible** source schema changes and stops ingestion safely rather than + silently corrupting downstream data. +- Supports **time travel / restore** — reconstructing a prior state from the warehouse. +- Mirrors source-side **validations** (system + business rules) in the warehouse. +- Publishes both lake and warehouse datasets to a **catalog** so they're discoverable. + +Correctness, explicit reasoning about failure modes, and clean layering matter more than +breadth or polish. I'm scoping this to a single well-reasoned domain rather than trying to +cover many edge cases shallowly. + +## 2. Chosen Domain + +**Wallet / Payments / Transfers** — realistic transactional shape, natural strong/weak entity +split, and gives me meaningful business validations to mirror (non-negative balances, valid +status transitions, settlement-after-creation ordering). + +Planned source schema (5 tables): + +| Table | Type | Notes | +|---|---|---| +| `customer` | Strong entity | id, name, email, created_at | +| `wallet` | Strong entity | id, customer_id (FK), currency, status (enum), created_at | +| `transfer` | Strong entity | id, source_wallet_id (FK), dest_wallet_id (FK), amount (decimal), status (enum), created_at, settled_at (nullable) | +| `wallet_balance_history` | Weak entity | id, wallet_id (FK), balance_after (decimal), recorded_at — depends on `wallet` for existence | +| `payment_attempt` | Weak entity | id, transfer_id (FK), attempt_no, status (enum), failure_reason (nullable), created_at | + +Indexes on FK columns and on `(wallet_id, recorded_at)` / `(transfer_id, attempt_no)` for +change-capture and lookup performance. Enum-like fields: `wallet.status`, +`transfer.status`, `payment_attempt.status`. Nullable fields: `transfer.settled_at`, +`payment_attempt.failure_reason`. + +## 3. Planned Tech Stack + +Kept intentionally light so the correctness story stays easy to follow: + +- **Source:** PostgreSQL (real constraints, real WAL available if I go the log-based route). +- **CDC:** Simulated/polling-based CDC using a `updated_at`/monotonic version column plus a + soft-delete flag, rather than full WAL/Debezium — documented as a simplification. If time + allows, I'll note what changes for a WAL-based (Debezium/logical replication) approach. +- **Lake:** Append-only Parquet (or JSON) change files, one record per captured change, with + operation type (`insert`/`update`/`delete`), before/after state, and a monotonic sequence + number for replay. +- **Warehouse:** DuckDB (or Postgres) holding both a current-state snapshot and an SCD2-style + history table per entity for time travel. +- **Validation:** Python/SQL assertions run as tests (pytest) against both source and + warehouse. +- **Catalog:** A minimal YAML/JSON registry file listing each lake/warehouse dataset, schema, + owner, and update cadence — with a note on what a production catalog (e.g. DataHub, Glue, + Unity Catalog) would add. + +## 4. CDC Strategy (planned) + +- Each source table gets a `updated_at` timestamp and `is_deleted` flag; a checkpoint table + tracks the last processed `(table, updated_at, id)` cursor per table. +- Each ingestion run pulls all rows with `updated_at > checkpoint`, writes them to the lake as + change records, then advances the checkpoint only after a successful lake write. +- **Duplicates:** lake records are keyed by `(table, pk, updated_at)`; replays are idempotent + because writes are upserts keyed on that tuple. +- **Deletes:** modeled as soft-deletes in source, captured as `op=delete` change records. +- **Restart/replay:** since the checkpoint only advances after a durable lake write, a crash + mid-run just re-reads the same window — no data loss, no duplication downstream. + +## 5. Schema Change Safety (planned) + +- Before each ingestion run, compare the live source schema (column names/types/nullability) + against a stored "expected schema" snapshot. +- Safe changes (new nullable column) are logged and ingestion continues. +- Breaking changes (dropped/renamed column, type change, tightened nullability, enum domain + shrink) cause ingestion to **halt** before writing anything for that run, with a clear + error/warning emitted (log line + non-zero exit / flagged status row). + +## 6. Time Travel / Restore (planned) + +- Lake is the source of truth for full history — any point-in-time state can be rebuilt by + replaying change records up to a timestamp. +- Warehouse keeps an SCD2 table per entity (`valid_from`, `valid_to`, `is_current`) so recent + history doesn't require a full lake replay; older restores fall back to lake replay into a + scratch table. + +## 7. Open Questions / Assumptions Going In + +- Simulated (polling-based) CDC is acceptable per the assignment text — I'll call out the + WAL-based production alternative explicitly rather than implement it. +- I'll implement SCD2 for at least `wallet` and `transfer` (the tables restore realistically + needs to touch); I may keep `payment_attempt` append-only-only if time is short, and will + note that tradeoff rather than hide it. +- AI usage: I'm using an AI assistant to help scaffold boilerplate (schema DDL, test + skeletons) — all logic, correctness reasoning, and final validation will be reviewed and + tested by me before submission, and I'll disclose specifics in the final PR description. + +## 8. Next Steps + +1. Source DDL + seed data. +2. CDC extraction script + checkpointing. +3. Lake writer (append-only change files). +4. Warehouse loader (snapshot + SCD2). +5. Schema-change detector. +6. Validation/test suite (Red → Blue → Green per requirement). +7. Catalog metadata file. +8. Final PR write-up covering all 7 required description points. \ No newline at end of file diff --git a/submission/megha/catalog.yaml b/submission/megha/catalog.yaml new file mode 100644 index 0000000..9b25f53 --- /dev/null +++ b/submission/megha/catalog.yaml @@ -0,0 +1,161 @@ +# catalog.yaml +# Minimal dataset catalog for discoverability. In production this would be +# published to a real catalog tool (DataHub, AWS Glue Catalog, Unity +# Catalog) — this file captures the same information manually, scoped to +# the datasets this assignment produces. + +datasets: + + # ------------------------------------------------------------------ + # LAKE — append-only change history, one JSON-lines file per source + # table. Full replay history; source of truth for time travel beyond + # what the warehouse's SCD2 tables retain. + # ------------------------------------------------------------------ + + - name: lake.customer + layer: lake + location: lake/customer.jsonl + format: json_lines + owner: data-platform-team + source_table: public.customer + update_cadence: on-demand (triggered by cdc_extract.py run; intended as scheduled/polling in production) + schema: + - {field: table, type: string, description: "source table name"} + - {field: op, type: string, description: "upsert | delete"} + - {field: pk, type: bigint, description: "primary key of source row"} + - {field: updated_at, type: timestamp, description: "source row's updated_at at capture time"} + - {field: captured_at, type: timestamp, description: "when this change record was written"} + - {field: data, type: object, description: "full row payload at time of capture"} + notes: > + Append-only. Never mutated or compacted in this assignment; a + production version would periodically compact/partition by date. + + - name: lake.wallet + layer: lake + location: lake/wallet.jsonl + format: json_lines + owner: data-platform-team + source_table: public.wallet + update_cadence: on-demand (triggered by cdc_extract.py run) + notes: Same record shape as lake.customer (table/op/pk/updated_at/captured_at/data); includes wallet.status and is_deleted. + + - name: lake.transfer + layer: lake + location: lake/transfer.jsonl + format: json_lines + owner: data-platform-team + source_table: public.transfer + update_cadence: on-demand (triggered by cdc_extract.py run) + notes: Same record shape as lake.customer; includes transfer.status, amount, settled_at. + + - name: lake.wallet_balance_history + layer: lake + location: lake/wallet_balance_history.jsonl + format: json_lines + owner: data-platform-team + source_table: public.wallet_balance_history + update_cadence: on-demand (triggered by cdc_extract.py run) + notes: Weak entity; depends on wallet. Append-only balance snapshots. + + - name: lake.payment_attempt + layer: lake + location: lake/payment_attempt.jsonl + format: json_lines + owner: data-platform-team + source_table: public.payment_attempt + update_cadence: on-demand (triggered by cdc_extract.py run) + notes: Weak entity; depends on transfer. + + # ------------------------------------------------------------------ + # WAREHOUSE — current-state snapshots (1 row per entity) + SCD2 + # history for wallet/transfer. Loaded from the lake by + # warehouse_load.py, idempotently. + # ------------------------------------------------------------------ + + - name: warehouse.customer + layer: warehouse + location: postgres://wallet_db/warehouse.customer + format: postgres_table + owner: data-platform-team + loaded_from: lake.customer + update_cadence: on-demand (triggered by warehouse_load.py run) + grain: current state, 1 row per customer.id + notes: Snapshot only — no history table for this entity in this assignment. + + - name: warehouse.wallet + layer: warehouse + location: postgres://wallet_db/warehouse.wallet + format: postgres_table + owner: data-platform-team + loaded_from: lake.wallet + update_cadence: on-demand (triggered by warehouse_load.py run) + grain: current state, 1 row per wallet.id + + - name: warehouse.wallet_history + layer: warehouse + location: postgres://wallet_db/warehouse.wallet_history + format: postgres_table + owner: data-platform-team + loaded_from: lake.wallet + update_cadence: on-demand (triggered by warehouse_load.py run) + grain: SCD2 — one row per (wallet.id, valid_from) state change + notes: > + Supports time travel: state as of any timestamp T is the row where + valid_from <= T AND (valid_to > T OR valid_to IS NULL). Limited by + polling-CDC granularity — see APPROACH.md for known gap on + intermediate-state loss between polls. + + - name: warehouse.transfer + layer: warehouse + location: postgres://wallet_db/warehouse.transfer + format: postgres_table + owner: data-platform-team + loaded_from: lake.transfer + update_cadence: on-demand (triggered by warehouse_load.py run) + grain: current state, 1 row per transfer.id + + - name: warehouse.transfer_history + layer: warehouse + location: postgres://wallet_db/warehouse.transfer_history + format: postgres_table + owner: data-platform-team + loaded_from: lake.transfer + update_cadence: on-demand (triggered by warehouse_load.py run) + grain: SCD2 — one row per (transfer.id, valid_from) state change + + - name: warehouse.wallet_balance_history + layer: warehouse + location: postgres://wallet_db/warehouse.wallet_balance_history + format: postgres_table + owner: data-platform-team + loaded_from: lake.wallet_balance_history + update_cadence: on-demand (triggered by warehouse_load.py run) + grain: current state, 1 row per wallet_balance_history.id (append-only by nature) + + - name: warehouse.payment_attempt + layer: warehouse + location: postgres://wallet_db/warehouse.payment_attempt + format: postgres_table + owner: data-platform-team + loaded_from: lake.payment_attempt + update_cadence: on-demand (triggered by warehouse_load.py run) + grain: current state, 1 row per payment_attempt.id + + # ------------------------------------------------------------------ + # OPERATIONAL — supporting tables (checkpoints, load logs). Listed for + # completeness/discoverability, not treated as analytical datasets. + # ------------------------------------------------------------------ + + - name: public.cdc_checkpoint + layer: operational + location: postgres://wallet_db/public.cdc_checkpoint + format: postgres_table + owner: data-platform-team + notes: Tracks last-processed (updated_at, id) cursor per source table for the CDC extractor. + + - name: warehouse.load_log + layer: operational + location: postgres://wallet_db/warehouse.load_log + format: postgres_table + owner: data-platform-team + notes: Tracks (table, pk, updated_at) already loaded into the warehouse, for idempotent replay. diff --git a/submission/megha/cdc_extract.py b/submission/megha/cdc_extract.py new file mode 100644 index 0000000..c625529 --- /dev/null +++ b/submission/megha/cdc_extract.py @@ -0,0 +1,154 @@ +#!/usr/bin/env python3 +""" +cdc_extract.py — polling-based CDC extractor. + +For each source table, pulls all rows with updated_at > last checkpoint +(using (updated_at, id) as a stable cursor to avoid skipping same-timestamp +rows), appends them as change records to the lake (JSON lines, one file per +table), and only then advances the checkpoint in Postgres. + +Idempotency: each lake record is keyed by (table, id, updated_at). Re-running +against the same window re-writes the same records rather than duplicating +downstream state — the lake file is append-only, but downstream consumers +(warehouse loader) de-dupe on that key. + +Usage: + python3 cdc_extract.py +""" + +import json +import os +from datetime import datetime, timezone + +import psycopg2 +import psycopg2.extras + +DB_CONFIG = dict( + host="localhost", + port=5432, + dbname="wallet_db", + user="postgres", + password="postgres", +) + +LAKE_DIR = "lake" +TABLES = [ + "customer", + "wallet", + "transfer", + "wallet_balance_history", + "payment_attempt", +] + + +def get_checkpoint(cur, table): + cur.execute( + "SELECT last_updated_at, last_id FROM cdc_checkpoint WHERE table_name = %s", + (table,), + ) + row = cur.fetchone() + return row["last_updated_at"], row["last_id"] + + +def advance_checkpoint(cur, table, last_updated_at, last_id): + cur.execute( + """ + UPDATE cdc_checkpoint + SET last_updated_at = %s, last_id = %s + WHERE table_name = %s + """, + (last_updated_at, last_id, table), + ) + + +def extract_table(conn, table): + """ + Returns the list of changed rows (as dicts) for this table since the + last checkpoint, and the new (updated_at, id) cursor to advance to. + Uses a compound cursor (updated_at, id) so multiple rows sharing the + same updated_at timestamp aren't skipped on the next run. + """ + with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + last_updated_at, last_id = get_checkpoint(cur, table) + + cur.execute( + f""" + SELECT * FROM {table} + WHERE (updated_at, id) > (%s, %s) + ORDER BY updated_at ASC, id ASC + """, + (last_updated_at, last_id), + ) + rows = cur.fetchall() + + if not rows: + return [], None + + new_cursor = (rows[-1]["updated_at"], rows[-1]["id"]) + return rows, new_cursor + + +def to_change_record(table, row): + op = "delete" if row.get("is_deleted") else "upsert" + # Serialize non-JSON-native types (datetime, Decimal) to strings. + payload = { + k: (str(v) if not isinstance(v, (str, int, float, bool, type(None))) else v) + for k, v in row.items() + } + return { + "table": table, + "op": op, + "pk": row["id"], + "updated_at": str(row["updated_at"]), + "captured_at": datetime.now(timezone.utc).isoformat(), + "data": payload, + } + + +def write_lake_records(table, records): + os.makedirs(LAKE_DIR, exist_ok=True) + path = os.path.join(LAKE_DIR, f"{table}.jsonl") + with open(path, "a") as f: + for rec in records: + f.write(json.dumps(rec) + "\n") + return path + + +def run(): + conn = psycopg2.connect(**DB_CONFIG) + conn.autocommit = False + total_changes = 0 + + try: + for table in TABLES: + rows, new_cursor = extract_table(conn, table) + + if not rows: + print(f"[{table}] no changes") + continue + + records = [to_change_record(table, row) for row in rows] + path = write_lake_records(table, records) + + # Only advance the checkpoint AFTER the lake write succeeds, + # and commit both in the same transaction as the lake file + # write already being on disk — this makes a crash here safe + # to just re-run (lake write is append-only + de-duped by + # downstream loaders on (table, pk, updated_at)). + with conn.cursor() as cur: + advance_checkpoint(cur, table, new_cursor[0], new_cursor[1]) + conn.commit() + + total_changes += len(records) + print(f"[{table}] captured {len(records)} change(s) -> {path}") + + print(f"\nDone. {total_changes} total change record(s) captured this run.") + except Exception: + conn.rollback() + raise + finally: + conn.close() + + +if __name__ == "__main__": + run() diff --git a/submission/megha/db/mutations.sql b/submission/megha/db/mutations.sql new file mode 100644 index 0000000..d2d1a61 --- /dev/null +++ b/submission/megha/db/mutations.sql @@ -0,0 +1,33 @@ +-- ============================================================ +-- Mutations: simulates a "next batch" of real-world activity. +-- Run this AFTER the first CDC extraction pass, so you have +-- something new for the second pass to pick up. +-- ============================================================ + +-- Update: pending transfer (id=2) becomes settled +UPDATE transfer +SET status = 'settled', settled_at = now(), updated_at = now() +WHERE id = 2; + +UPDATE payment_attempt +SET status = 'succeeded', updated_at = now() +WHERE transfer_id = 2 AND attempt_no = 1; + +INSERT INTO wallet_balance_history (wallet_id, balance_after) VALUES + (3, 8800.00), + (1, 5700.00); + +-- Insert: a new customer + wallet + transfer that fails +INSERT INTO customer (name, email) VALUES ('Rohit Verma', 'rohit.verma@example.com'); +INSERT INTO wallet (customer_id, currency, status) VALUES (4, 'INR', 'active'); + +INSERT INTO transfer (source_wallet_id, dest_wallet_id, amount, status) VALUES + (2, 4, 250.00, 'failed'); + +INSERT INTO payment_attempt (transfer_id, attempt_no, status, failure_reason) VALUES + (3, 1, 'failed', 'insufficient_funds'); + +-- Soft-delete: wallet 4 gets closed and marked deleted (edge case for CDC) +UPDATE wallet +SET status = 'closed', is_deleted = true, updated_at = now() +WHERE id = 4; diff --git a/submission/megha/db/schema.sql b/submission/megha/db/schema.sql new file mode 100644 index 0000000..3bd72d7 --- /dev/null +++ b/submission/megha/db/schema.sql @@ -0,0 +1,104 @@ +-- ============================================================ +-- Source schema: wallet / payments / transfers domain +-- Includes updated_at + is_deleted on every table to support +-- polling-based CDC (see APPROACH.md). +-- ============================================================ + +CREATE TYPE wallet_status AS ENUM ('active', 'frozen', 'closed'); +CREATE TYPE transfer_status AS ENUM ('pending', 'settled', 'failed', 'reversed'); +CREATE TYPE payment_attempt_status AS ENUM ('initiated', 'succeeded', 'failed'); + +-- --------------------------------------------- +-- customer (strong entity) +-- --------------------------------------------- +CREATE TABLE customer ( + id BIGSERIAL PRIMARY KEY, + name TEXT NOT NULL, + email TEXT NOT NULL UNIQUE, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + is_deleted BOOLEAN NOT NULL DEFAULT false +); + +-- --------------------------------------------- +-- wallet (strong entity) +-- --------------------------------------------- +CREATE TABLE wallet ( + id BIGSERIAL PRIMARY KEY, + customer_id BIGINT NOT NULL REFERENCES customer(id), + currency CHAR(3) NOT NULL, + status wallet_status NOT NULL DEFAULT 'active', + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + is_deleted BOOLEAN NOT NULL DEFAULT false +); + +CREATE INDEX idx_wallet_customer_id ON wallet(customer_id); +CREATE INDEX idx_wallet_updated_at ON wallet(updated_at); + +-- --------------------------------------------- +-- transfer (strong entity) +-- --------------------------------------------- +CREATE TABLE transfer ( + id BIGSERIAL PRIMARY KEY, + source_wallet_id BIGINT NOT NULL REFERENCES wallet(id), + dest_wallet_id BIGINT NOT NULL REFERENCES wallet(id), + amount NUMERIC(18, 2) NOT NULL CHECK (amount > 0), + status transfer_status NOT NULL DEFAULT 'pending', + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + settled_at TIMESTAMPTZ, + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + is_deleted BOOLEAN NOT NULL DEFAULT false, + CHECK (source_wallet_id <> dest_wallet_id), + CHECK (settled_at IS NULL OR settled_at >= created_at) +); + +CREATE INDEX idx_transfer_source_wallet ON transfer(source_wallet_id); +CREATE INDEX idx_transfer_dest_wallet ON transfer(dest_wallet_id); +CREATE INDEX idx_transfer_updated_at ON transfer(updated_at); + +-- --------------------------------------------- +-- wallet_balance_history (weak entity, depends on wallet) +-- --------------------------------------------- +CREATE TABLE wallet_balance_history ( + id BIGSERIAL PRIMARY KEY, + wallet_id BIGINT NOT NULL REFERENCES wallet(id) ON DELETE CASCADE, + balance_after NUMERIC(18, 2) NOT NULL, + recorded_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + is_deleted BOOLEAN NOT NULL DEFAULT false +); + +CREATE INDEX idx_wbh_wallet_recorded ON wallet_balance_history(wallet_id, recorded_at); +CREATE INDEX idx_wbh_updated_at ON wallet_balance_history(updated_at); + +-- --------------------------------------------- +-- payment_attempt (weak entity, depends on transfer) +-- --------------------------------------------- +CREATE TABLE payment_attempt ( + id BIGSERIAL PRIMARY KEY, + transfer_id BIGINT NOT NULL REFERENCES transfer(id) ON DELETE CASCADE, + attempt_no INT NOT NULL, + status payment_attempt_status NOT NULL DEFAULT 'initiated', + failure_reason TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + is_deleted BOOLEAN NOT NULL DEFAULT false, + UNIQUE (transfer_id, attempt_no) +); + +CREATE INDEX idx_pa_transfer_attempt ON payment_attempt(transfer_id, attempt_no); +CREATE INDEX idx_pa_updated_at ON payment_attempt(updated_at); + +-- --------------------------------------------- +-- CDC checkpoint table (tracks last-processed cursor per source table) +-- --------------------------------------------- +CREATE TABLE cdc_checkpoint ( + table_name TEXT PRIMARY KEY, + last_updated_at TIMESTAMPTZ NOT NULL DEFAULT 'epoch', + last_id BIGINT NOT NULL DEFAULT 0 +); + +INSERT INTO cdc_checkpoint (table_name) VALUES + ('customer'), ('wallet'), ('transfer'), + ('wallet_balance_history'), ('payment_attempt'); \ No newline at end of file diff --git a/submission/megha/db/seed.sql b/submission/megha/db/seed.sql new file mode 100644 index 0000000..4a2443d --- /dev/null +++ b/submission/megha/db/seed.sql @@ -0,0 +1,41 @@ +-- ============================================================ +-- Seed data: initial state +-- Run this once after schema.sql to populate baseline rows. +-- ============================================================ + +-- Customers +INSERT INTO customer (name, email) VALUES + ('Asha Rao', 'asha.rao@example.com'), + ('Vikram Shah', 'vikram.shah@example.com'), + ('Priya Nair', 'priya.nair@example.com'); + +-- Wallets (one each to start, INR) +INSERT INTO wallet (customer_id, currency, status) VALUES + (1, 'INR', 'active'), + (2, 'INR', 'active'), + (3, 'INR', 'active'); + +-- Initial balance history (opening balances) +INSERT INTO wallet_balance_history (wallet_id, balance_after) VALUES + (1, 5000.00), + (2, 3000.00), + (3, 10000.00); + +-- A settled transfer: wallet 1 -> wallet 2 +INSERT INTO transfer (source_wallet_id, dest_wallet_id, amount, status, settled_at) VALUES + (1, 2, 500.00, 'settled', now()); + +INSERT INTO payment_attempt (transfer_id, attempt_no, status) VALUES + (1, 1, 'succeeded'); + +-- Reflect the settled transfer in balance history +INSERT INTO wallet_balance_history (wallet_id, balance_after) VALUES + (1, 4500.00), + (2, 3500.00); + +-- A pending transfer: wallet 3 -> wallet 1 +INSERT INTO transfer (source_wallet_id, dest_wallet_id, amount, status) VALUES + (3, 1, 1200.00, 'pending'); + +INSERT INTO payment_attempt (transfer_id, attempt_no, status) VALUES + (2, 1, 'initiated'); diff --git a/submission/megha/db/warehouse_schema.sql b/submission/megha/db/warehouse_schema.sql new file mode 100644 index 0000000..9754508 --- /dev/null +++ b/submission/megha/db/warehouse_schema.sql @@ -0,0 +1,107 @@ +-- ============================================================ +-- Warehouse schema: current-state snapshots + SCD2 history +-- for wallet and transfer (the two entities restore/time-travel +-- realistically needs). Other entities are snapshot-only. +-- ============================================================ + +CREATE SCHEMA IF NOT EXISTS warehouse; + +-- --------------------------------------------- +-- Current-state snapshot tables (1 row per entity, latest state) +-- --------------------------------------------- +CREATE TABLE warehouse.customer ( + id BIGINT PRIMARY KEY, + name TEXT NOT NULL, + email TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL, + updated_at TIMESTAMPTZ NOT NULL, + is_deleted BOOLEAN NOT NULL +); + +CREATE TABLE warehouse.wallet ( + id BIGINT PRIMARY KEY, + customer_id BIGINT NOT NULL, + currency TEXT NOT NULL, + status TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL, + updated_at TIMESTAMPTZ NOT NULL, + is_deleted BOOLEAN NOT NULL +); + +CREATE TABLE warehouse.transfer ( + id BIGINT PRIMARY KEY, + source_wallet_id BIGINT NOT NULL, + dest_wallet_id BIGINT NOT NULL, + amount NUMERIC(18, 2) NOT NULL, + status TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL, + settled_at TIMESTAMPTZ, + updated_at TIMESTAMPTZ NOT NULL, + is_deleted BOOLEAN NOT NULL +); + +CREATE TABLE warehouse.wallet_balance_history ( + id BIGINT PRIMARY KEY, + wallet_id BIGINT NOT NULL, + balance_after NUMERIC(18, 2) NOT NULL, + recorded_at TIMESTAMPTZ NOT NULL, + updated_at TIMESTAMPTZ NOT NULL, + is_deleted BOOLEAN NOT NULL +); + +CREATE TABLE warehouse.payment_attempt ( + id BIGINT PRIMARY KEY, + transfer_id BIGINT NOT NULL, + attempt_no INT NOT NULL, + status TEXT NOT NULL, + failure_reason TEXT, + created_at TIMESTAMPTZ NOT NULL, + updated_at TIMESTAMPTZ NOT NULL, + is_deleted BOOLEAN NOT NULL +); + +-- --------------------------------------------- +-- SCD2 history tables — wallet and transfer only. +-- valid_to = NULL means "currently valid" (is_current = true). +-- --------------------------------------------- +CREATE TABLE warehouse.wallet_history ( + surrogate_id BIGSERIAL PRIMARY KEY, + id BIGINT NOT NULL, + customer_id BIGINT NOT NULL, + currency TEXT NOT NULL, + status TEXT NOT NULL, + is_deleted BOOLEAN NOT NULL, + valid_from TIMESTAMPTZ NOT NULL, + valid_to TIMESTAMPTZ, + is_current BOOLEAN NOT NULL DEFAULT true +); + +CREATE INDEX idx_wallet_history_id_current ON warehouse.wallet_history(id, is_current); + +CREATE TABLE warehouse.transfer_history ( + surrogate_id BIGSERIAL PRIMARY KEY, + id BIGINT NOT NULL, + source_wallet_id BIGINT NOT NULL, + dest_wallet_id BIGINT NOT NULL, + amount NUMERIC(18, 2) NOT NULL, + status TEXT NOT NULL, + settled_at TIMESTAMPTZ, + is_deleted BOOLEAN NOT NULL, + valid_from TIMESTAMPTZ NOT NULL, + valid_to TIMESTAMPTZ, + is_current BOOLEAN NOT NULL DEFAULT true +); + +CREATE INDEX idx_transfer_history_id_current ON warehouse.transfer_history(id, is_current); + +-- --------------------------------------------- +-- Loader watermark: tracks last lake record processed per table, +-- keyed by (table, pk, updated_at) so replays are idempotent. +-- --------------------------------------------- +CREATE TABLE warehouse.load_log ( + table_name TEXT NOT NULL, + pk BIGINT NOT NULL, + updated_at TIMESTAMPTZ NOT NULL, + loaded_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (table_name, pk, updated_at) +); diff --git a/submission/megha/schema_guard.py b/submission/megha/schema_guard.py new file mode 100644 index 0000000..adf4220 --- /dev/null +++ b/submission/megha/schema_guard.py @@ -0,0 +1,172 @@ +#!/usr/bin/env python3 +""" +schema_guard.py — detects source schema drift before CDC extraction runs. + +On first run (no stored snapshot), captures the current schema of every +CDC-tracked table as the "expected" baseline and saves it to +schema_snapshot.json. + +On every subsequent run, compares the live schema against that baseline: + - SAFE changes (new nullable column added) are logged as warnings and + ingestion is allowed to proceed. The baseline is updated to include + the new column so it's no longer flagged next time. + - BREAKING changes (column dropped/renamed, type changed, nullable + column made NOT NULL, column narrowed) cause this script to exit + with a non-zero status, and cdc_extract.py should not be invoked. + +Usage: + python3 schema_guard.py # check only, exit 1 on breaking change + python3 schema_guard.py --init # (re)initialize the baseline snapshot +""" + +import json +import sys + +import psycopg2 +import psycopg2.extras + +DB_CONFIG = dict( + host="localhost", + port=5432, + dbname="wallet_db", + user="postgres", + password="postgres", +) + +TABLES = [ + "customer", + "wallet", + "transfer", + "wallet_balance_history", + "payment_attempt", +] + +SNAPSHOT_PATH = "schema_snapshot.json" + + +def get_live_schema(conn): + """Returns {table: {column: {type, nullable}}} for all tracked tables.""" + schema = {} + with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + for table in TABLES: + cur.execute( + """ + SELECT column_name, data_type, is_nullable + FROM information_schema.columns + WHERE table_schema = 'public' AND table_name = %s + """, + (table,), + ) + cols = { + row["column_name"]: { + "type": row["data_type"], + "nullable": row["is_nullable"] == "YES", + } + for row in cur.fetchall() + } + schema[table] = cols + return schema + + +def load_baseline(): + try: + with open(SNAPSHOT_PATH) as f: + return json.load(f) + except FileNotFoundError: + return None + + +def save_baseline(schema): + with open(SNAPSHOT_PATH, "w") as f: + json.dump(schema, f, indent=2, sort_keys=True) + + +def diff_schema(baseline, live): + """ + Returns (breaking_changes, safe_changes), each a list of human-readable + strings. + """ + breaking = [] + safe = [] + + for table, expected_cols in baseline.items(): + live_cols = live.get(table) + if live_cols is None: + breaking.append(f"{table}: table dropped entirely") + continue + + for col, expected in expected_cols.items(): + live_col = live_cols.get(col) + if live_col is None: + breaking.append(f"{table}.{col}: column dropped") + continue + if live_col["type"] != expected["type"]: + breaking.append( + f"{table}.{col}: type changed " + f"({expected['type']} -> {live_col['type']})" + ) + if expected["nullable"] and not live_col["nullable"]: + breaking.append( + f"{table}.{col}: nullability tightened " + f"(was nullable, now NOT NULL)" + ) + + for col in live_cols: + if col not in expected_cols: + safe.append(f"{table}.{col}: new column added") + + for table in live: + if table not in baseline: + safe.append(f"{table}: new table added") + + return breaking, safe + + +def run(init=False): + conn = psycopg2.connect(**DB_CONFIG) + try: + live_schema = get_live_schema(conn) + finally: + conn.close() + + if init: + save_baseline(live_schema) + print(f"Baseline schema snapshot written to {SNAPSHOT_PATH}.") + return 0 + + baseline = load_baseline() + if baseline is None: + print("No baseline found — initializing one now (first run).") + save_baseline(live_schema) + return 0 + + breaking, safe = diff_schema(baseline, live_schema) + + if safe: + print("Safe schema changes detected (ingestion will proceed):") + for s in safe: + print(f" - {s}") + # Merge new columns/tables into the baseline so they're not + # re-flagged on the next run. + save_baseline(live_schema) + + if breaking: + print("\nBREAKING schema change(s) detected — halting ingestion:") + for b in breaking: + print(f" - {b}") + print( + "\nCDC extraction has NOT been run. Resolve the schema change, " + "then re-run with --init to accept the new baseline (only after " + "downstream lake/warehouse mappings have been updated to match)." + ) + return 1 + + if not safe: + print("No schema changes detected. Safe to proceed.") + + return 0 + + +if __name__ == "__main__": + init = "--init" in sys.argv + sys.exit(run(init=init)) diff --git a/submission/megha/schema_snapshot.json b/submission/megha/schema_snapshot.json new file mode 100644 index 0000000..cc49e2e --- /dev/null +++ b/submission/megha/schema_snapshot.json @@ -0,0 +1,156 @@ +{ + "customer": { + "created_at": { + "nullable": false, + "type": "timestamp with time zone" + }, + "email": { + "nullable": false, + "type": "text" + }, + "id": { + "nullable": false, + "type": "bigint" + }, + "is_deleted": { + "nullable": false, + "type": "boolean" + }, + "name": { + "nullable": false, + "type": "text" + }, + "updated_at": { + "nullable": false, + "type": "timestamp with time zone" + } + }, + "payment_attempt": { + "attempt_no": { + "nullable": false, + "type": "integer" + }, + "created_at": { + "nullable": false, + "type": "timestamp with time zone" + }, + "failure_reason": { + "nullable": true, + "type": "text" + }, + "id": { + "nullable": false, + "type": "bigint" + }, + "is_deleted": { + "nullable": false, + "type": "boolean" + }, + "status": { + "nullable": false, + "type": "USER-DEFINED" + }, + "transfer_id": { + "nullable": false, + "type": "bigint" + }, + "updated_at": { + "nullable": false, + "type": "timestamp with time zone" + } + }, + "transfer": { + "amount": { + "nullable": false, + "type": "numeric" + }, + "created_at": { + "nullable": false, + "type": "timestamp with time zone" + }, + "dest_wallet_id": { + "nullable": false, + "type": "bigint" + }, + "id": { + "nullable": false, + "type": "bigint" + }, + "is_deleted": { + "nullable": false, + "type": "boolean" + }, + "settled_at": { + "nullable": true, + "type": "timestamp with time zone" + }, + "source_wallet_id": { + "nullable": false, + "type": "bigint" + }, + "status": { + "nullable": false, + "type": "USER-DEFINED" + }, + "updated_at": { + "nullable": false, + "type": "timestamp with time zone" + } + }, + "wallet": { + "created_at": { + "nullable": false, + "type": "timestamp with time zone" + }, + "currency": { + "nullable": false, + "type": "character" + }, + "customer_id": { + "nullable": false, + "type": "bigint" + }, + "id": { + "nullable": false, + "type": "bigint" + }, + "is_deleted": { + "nullable": false, + "type": "boolean" + }, + "status": { + "nullable": false, + "type": "USER-DEFINED" + }, + "updated_at": { + "nullable": false, + "type": "timestamp with time zone" + } + }, + "wallet_balance_history": { + "balance_after": { + "nullable": false, + "type": "numeric" + }, + "id": { + "nullable": false, + "type": "bigint" + }, + "is_deleted": { + "nullable": false, + "type": "boolean" + }, + "recorded_at": { + "nullable": false, + "type": "timestamp with time zone" + }, + "updated_at": { + "nullable": false, + "type": "timestamp with time zone" + }, + "wallet_id": { + "nullable": false, + "type": "bigint" + } + } +} \ No newline at end of file diff --git a/submission/megha/test_pipeline.py b/submission/megha/test_pipeline.py new file mode 100644 index 0000000..fe232b5 --- /dev/null +++ b/submission/megha/test_pipeline.py @@ -0,0 +1,266 @@ +""" +test_pipeline.py — validation suite for the CDC lakehouse pipeline. + +Covers the categories required by the assignment: + 1. Modeling/constraints — source-side business rules are enforced. + 2. CDC correctness — insert/update/delete capture, idempotent replay. + 3. Warehouse correctness — snapshot matches source, SCD2 history is sane. + 4. Schema-change safety — breaking changes are detected and halt ingestion. + +Assumes: Postgres running (docker container `cdc-postgres`), schema + +warehouse_schema already applied, and seed.sql already loaded at least +once. Some tests insert/mutate their own throwaway rows (cleaned up via +rollback) rather than depending on exact seed state, so this suite is +safe to re-run. + +Run with: + pip3 install pytest psycopg2-binary + pytest submission/megha/test_pipeline.py -v +""" + +import json +import os +import subprocess +import sys + +import psycopg2 +import psycopg2.extras +import pytest + +DB_CONFIG = dict( + host="localhost", port=5432, dbname="wallet_db", + user="postgres", password="postgres", +) + +# The .py scripts (cdc_extract.py, warehouse_load.py, schema_guard.py) live +# alongside this test file... +SCRIPTS_DIR = os.path.dirname(os.path.abspath(__file__)) + +# ...but they're always invoked with the repo root as cwd (e.g. +# `python3 submission/megha/cdc_extract.py` run from ~/data-assignments), +# so that's where they write lake/ and schema_snapshot.json. Tests must +# use that same root for both finding data files and setting subprocess cwd. +REPO_ROOT = os.path.abspath(os.path.join(SCRIPTS_DIR, "..", "..")) + + +def script_path(name): + return os.path.join(SCRIPTS_DIR, name) + + +@pytest.fixture +def conn(): + c = psycopg2.connect(**DB_CONFIG) + c.autocommit = False + yield c + c.rollback() # never persist test-only mutations + c.close() + + +@pytest.fixture +def cur(conn): + with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as c: + yield c + + +# ------------------------------------------------------------------ +# 1. Modeling / constraints +# ------------------------------------------------------------------ + +class TestSourceConstraints: + def test_transfer_amount_must_be_positive(self, conn, cur): + with pytest.raises(psycopg2.errors.CheckViolation): + cur.execute( + "INSERT INTO transfer (source_wallet_id, dest_wallet_id, amount) " + "VALUES (1, 2, -50.00)" + ) + conn.rollback() + + def test_transfer_cannot_self_reference(self, conn, cur): + with pytest.raises(psycopg2.errors.CheckViolation): + cur.execute( + "INSERT INTO transfer (source_wallet_id, dest_wallet_id, amount) " + "VALUES (1, 1, 50.00)" + ) + conn.rollback() + + def test_settled_at_cannot_precede_created_at(self, conn, cur): + with pytest.raises(psycopg2.errors.CheckViolation): + cur.execute( + "INSERT INTO transfer " + "(source_wallet_id, dest_wallet_id, amount, created_at, settled_at) " + "VALUES (1, 2, 50.00, now(), now() - interval '1 day')" + ) + conn.rollback() + + def test_customer_email_must_be_unique(self, conn, cur): + cur.execute("SELECT email FROM customer LIMIT 1") + existing_email = cur.fetchone()["email"] + with pytest.raises(psycopg2.errors.UniqueViolation): + cur.execute( + "INSERT INTO customer (name, email) VALUES ('Dup', %s)", + (existing_email,), + ) + conn.rollback() + + def test_payment_attempt_no_must_be_unique_per_transfer(self, conn, cur): + cur.execute("SELECT id FROM transfer LIMIT 1") + transfer_id = cur.fetchone()["id"] + cur.execute( + "SELECT attempt_no FROM payment_attempt WHERE transfer_id = %s LIMIT 1", + (transfer_id,), + ) + row = cur.fetchone() + if row is None: + pytest.skip("no existing payment_attempt to test uniqueness against") + with pytest.raises(psycopg2.errors.UniqueViolation): + cur.execute( + "INSERT INTO payment_attempt (transfer_id, attempt_no, status) " + "VALUES (%s, %s, 'initiated')", + (transfer_id, row["attempt_no"]), + ) + conn.rollback() + + +# ------------------------------------------------------------------ +# 2. CDC correctness +# ------------------------------------------------------------------ + +class TestCdcCorrectness: + def test_extractor_is_idempotent_with_no_new_data(self): + """Running the extractor twice in a row with no source changes + should capture 0 records on the second run.""" + result = subprocess.run( + [sys.executable, script_path("cdc_extract.py")], + cwd=REPO_ROOT, capture_output=True, text=True, + ) + assert result.returncode == 0, result.stderr + assert "Done. 0 total change record(s)" in result.stdout, ( + "Expected no new changes on a repeat run with no source " + f"mutations; got:\n{result.stdout}" + ) + + def test_lake_records_have_required_fields(self): + path = os.path.join(REPO_ROOT, "lake", "wallet.jsonl") + if not os.path.exists(path): + pytest.skip("lake/wallet.jsonl not present — run cdc_extract.py first") + with open(path) as f: + lines = f.readlines() + assert len(lines) > 0 + for line in lines: + record = json.loads(line) + for field in ("table", "op", "pk", "updated_at", "captured_at", "data"): + assert field in record, f"missing field {field} in {record}" + assert record["op"] in ("upsert", "delete") + + def test_soft_deleted_row_captured_as_delete_op(self): + path = os.path.join(REPO_ROOT, "lake", "wallet.jsonl") + if not os.path.exists(path): + pytest.skip("lake/wallet.jsonl not present — run cdc_extract.py first") + with open(path) as f: + records = [json.loads(line) for line in f] + deleted = [r for r in records if r["data"].get("is_deleted") is True] + for r in deleted: + assert r["op"] == "delete", ( + f"row {r['pk']} has is_deleted=true but op={r['op']!r}" + ) + + +# ------------------------------------------------------------------ +# 3. Warehouse correctness +# ------------------------------------------------------------------ + +class TestWarehouseCorrectness: + def test_warehouse_wallet_count_matches_source(self, cur): + cur.execute("SELECT count(*) AS n FROM wallet") + source_count = cur.fetchone()["n"] + cur.execute("SELECT count(*) AS n FROM warehouse.wallet") + warehouse_count = cur.fetchone()["n"] + assert source_count == warehouse_count + + def test_warehouse_snapshot_matches_source_for_each_wallet(self, cur): + cur.execute("SELECT id, status, is_deleted FROM wallet") + source_rows = {r["id"]: (r["status"], r["is_deleted"]) for r in cur.fetchall()} + + cur.execute("SELECT id, status, is_deleted FROM warehouse.wallet") + warehouse_rows = {r["id"]: (r["status"], r["is_deleted"]) for r in cur.fetchall()} + + assert source_rows == warehouse_rows + + def test_scd2_history_has_exactly_one_current_row_per_entity(self, cur): + cur.execute( + """ + SELECT id, count(*) AS n + FROM warehouse.wallet_history + WHERE is_current = true + GROUP BY id + HAVING count(*) != 1 + """ + ) + violations = cur.fetchall() + assert violations == [], f"entities with != 1 current row: {violations}" + + def test_scd2_closed_rows_have_valid_to_set(self, cur): + cur.execute( + "SELECT id, surrogate_id FROM warehouse.transfer_history " + "WHERE is_current = false AND valid_to IS NULL" + ) + violations = cur.fetchall() + assert violations == [], f"closed rows missing valid_to: {violations}" + + def test_loader_is_idempotent(self): + """Running the loader twice with no new lake records should load 0.""" + result = subprocess.run( + [sys.executable, script_path("warehouse_load.py")], + cwd=REPO_ROOT, capture_output=True, text=True, + ) + assert result.returncode == 0, result.stderr + assert "Done. 0 total record(s) loaded" in result.stdout, ( + f"Expected idempotent no-op load; got:\n{result.stdout}" + ) + + +# ------------------------------------------------------------------ +# 4. Schema-change safety +# ------------------------------------------------------------------ + +class TestSchemaChangeSafety: + def test_no_breaking_change_exits_zero(self): + result = subprocess.run( + [sys.executable, script_path("schema_guard.py")], + cwd=REPO_ROOT, capture_output=True, text=True, + ) + assert result.returncode == 0, result.stderr + + def test_dropped_column_is_detected_as_breaking(self, conn, cur): + # Defensive cleanup: a previous failed run may have left this + # column behind before reaching its own cleanup step. + cur.execute("ALTER TABLE payment_attempt DROP COLUMN IF EXISTS temp_test_col") + conn.commit() + + # Add a throwaway nullable column, initialize baseline to include it, + # then drop it and confirm schema_guard flags it and exits non-zero. + cur.execute("ALTER TABLE payment_attempt ADD COLUMN temp_test_col TEXT") + conn.commit() + + init_result = subprocess.run( + [sys.executable, script_path("schema_guard.py"), "--init"], + cwd=REPO_ROOT, capture_output=True, text=True, + ) + assert init_result.returncode == 0, init_result.stderr + + cur.execute("ALTER TABLE payment_attempt DROP COLUMN temp_test_col") + conn.commit() + + check_result = subprocess.run( + [sys.executable, script_path("schema_guard.py")], + cwd=REPO_ROOT, capture_output=True, text=True, + ) + assert check_result.returncode == 1 + assert "temp_test_col" in check_result.stdout + assert "BREAKING" in check_result.stdout + + # Restore baseline to real schema so later test runs aren't affected. + subprocess.run( + [sys.executable, script_path("schema_guard.py"), "--init"], + cwd=REPO_ROOT, capture_output=True, text=True, + ) diff --git a/submission/megha/warehouse_load.py b/submission/megha/warehouse_load.py new file mode 100644 index 0000000..dc73d1a --- /dev/null +++ b/submission/megha/warehouse_load.py @@ -0,0 +1,178 @@ +#!/usr/bin/env python3 +""" +warehouse_load.py — loads lake change records into the warehouse. + +Reads each lake/{table}.jsonl file, and for every record not already in +warehouse.load_log (keyed by table+pk+updated_at, so replays are a no-op): + 1. Upserts the current-state snapshot table (warehouse.{table}). + 2. For wallet/transfer: closes out the previous SCD2 row (sets valid_to, + is_current=false) and inserts a new current row. + 3. Records the (table, pk, updated_at) in load_log. + +All three steps for a given record happen in one transaction, so a crash +mid-load never leaves the snapshot and history tables inconsistent. + +Usage: + python3 warehouse_load.py +""" + +import json +import os + +import psycopg2 +import psycopg2.extras + +DB_CONFIG = dict( + host="localhost", + port=5432, + dbname="wallet_db", + user="postgres", + password="postgres", +) + +LAKE_DIR = "lake" + +SNAPSHOT_COLUMNS = { + "customer": ["id", "name", "email", "created_at", "updated_at", "is_deleted"], + "wallet": ["id", "customer_id", "currency", "status", "created_at", "updated_at", "is_deleted"], + "transfer": ["id", "source_wallet_id", "dest_wallet_id", "amount", "status", + "created_at", "settled_at", "updated_at", "is_deleted"], + "wallet_balance_history": ["id", "wallet_id", "balance_after", "recorded_at", + "updated_at", "is_deleted"], + "payment_attempt": ["id", "transfer_id", "attempt_no", "status", "failure_reason", + "created_at", "updated_at", "is_deleted"], +} + +SCD2_TABLES = {"wallet", "transfer"} + +SCD2_COLUMNS = { + "wallet": ["id", "customer_id", "currency", "status", "is_deleted"], + "transfer": ["id", "source_wallet_id", "dest_wallet_id", "amount", "status", + "settled_at", "is_deleted"], +} + + +def already_loaded(cur, table, pk, updated_at): + cur.execute( + """ + SELECT 1 FROM warehouse.load_log + WHERE table_name = %s AND pk = %s AND updated_at = %s + """, + (table, pk, updated_at), + ) + return cur.fetchone() is not None + + +def upsert_snapshot(cur, table, data): + cols = SNAPSHOT_COLUMNS[table] + values = [data.get(c) for c in cols] + placeholders = ", ".join(["%s"] * len(cols)) + update_clause = ", ".join(f"{c} = EXCLUDED.{c}" for c in cols if c != "id") + cur.execute( + f""" + INSERT INTO warehouse.{table} ({", ".join(cols)}) + VALUES ({placeholders}) + ON CONFLICT (id) DO UPDATE SET {update_clause} + """, + values, + ) + + +def upsert_scd2(cur, table, data): + history_table = f"warehouse.{table}_history" + cols = SCD2_COLUMNS[table] + entity_id = data["id"] + valid_from = data["updated_at"] + + # Close out the current row, if one exists and actually differs. + cur.execute( + f"SELECT {', '.join(cols)} FROM {history_table} WHERE id = %s AND is_current = true", + (entity_id,), + ) + current = cur.fetchone() + + new_values = tuple(str(data.get(c)) for c in cols) + if current is not None and tuple(str(v) for v in current) == new_values: + # No actual change in tracked columns (e.g. metadata-only touch) — skip. + return + + if current is not None: + cur.execute( + f""" + UPDATE {history_table} + SET valid_to = %s, is_current = false + WHERE id = %s AND is_current = true + """, + (valid_from, entity_id), + ) + + values = [data.get(c) for c in cols] + [valid_from] + placeholders = ", ".join(["%s"] * len(cols)) + cur.execute( + f""" + INSERT INTO {history_table} ({", ".join(cols)}, valid_from, is_current) + VALUES ({placeholders}, %s, true) + """, + values, + ) + + +def load_table(conn, table): + path = os.path.join(LAKE_DIR, f"{table}.jsonl") + if not os.path.exists(path): + print(f"[{table}] no lake file, skipping") + return 0 + + loaded = 0 + with open(path) as f: + lines = f.readlines() + + with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + for line in lines: + record = json.loads(line) + data = record["data"] + pk = record["pk"] + updated_at = record["updated_at"] + + if already_loaded(cur, table, pk, updated_at): + continue + + upsert_snapshot(cur, table, data) + + if table in SCD2_TABLES: + upsert_scd2(cur, table, data) + + cur.execute( + """ + INSERT INTO warehouse.load_log (table_name, pk, updated_at) + VALUES (%s, %s, %s) + ON CONFLICT DO NOTHING + """, + (table, pk, updated_at), + ) + loaded += 1 + + conn.commit() + return loaded + + +def run(): + conn = psycopg2.connect(**DB_CONFIG) + conn.autocommit = False + total = 0 + + try: + for table in SNAPSHOT_COLUMNS: + n = load_table(conn, table) + print(f"[{table}] loaded {n} new record(s)") + total += n + print(f"\nDone. {total} total record(s) loaded into warehouse.") + except Exception: + conn.rollback() + raise + finally: + conn.close() + + +if __name__ == "__main__": + run()