Add initial approach doc for CDC lakehouse assignment

.gitignore

@@ -0,0 +1,4 @@
+.DS_Store
+__pycache__/
+.pytest_cache/
+*.pyc

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}}

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}}

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}}

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}}

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}}

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.