Add initial approach doc for CDC lakehouse assignment

[MeghaNandish]

CDC Lakehouse — Wallet/Payments Domain

Implements a small but complete CDC lakehouse pipeline: Postgres source → polling-based CDC → append-only JSON-lines lake → Postgres warehouse (current-state snapshot + SCD2 history) → schema-change guard → validation suite → dataset catalog.

All commands below assume repo root as cwd and the cdc-postgres Docker container running (docker run --name cdc-postgres -e POSTGRES_PASSWORD=postgres -e POSTGRES_DB=wallet_db -p 5432:5432 -d postgres:16).

How to run

pip3 install psycopg2-binary pytest
1. Source schema + seed data
docker cp submission/megha/db/schema.sql cdc-postgres:/schema.sql

docker exec -it cdc-postgres psql -U postgres -d wallet_db -f /schema.sql

docker cp submission/megha/db/seed.sql cdc-postgres:/seed.sql

docker exec -it cdc-postgres psql -U postgres -d wallet_db -f /seed.sql
2. Warehouse schema
docker cp submission/megha/db/warehouse_schema.sql cdc-postgres:/warehouse_schema.sql

docker exec -it cdc-postgres psql -U postgres -d wallet_db -f /warehouse_schema.sql
3. Schema baseline (before first CDC run)
python3 submission/megha/schema_guard.py --init
4. Run the pipeline
python3 submission/megha/cdc_extract.py

python3 submission/megha/warehouse_load.py
5. (Optional) simulate a second batch of activity, then re-run 3–4
docker cp submission/megha/db/mutations.sql cdc-postgres:/mutations.sql

docker exec -it cdc-postgres psql -U postgres -d wallet_db -f /mutations.sql

python3 submission/megha/schema_guard.py

python3 submission/megha/cdc_extract.py

python3 submission/megha/warehouse_load.py
6. Validation suite
python3 -m pytest submission/megha/test_pipeline.py -v

---

1. Source Schema

Domain: wallet / payments / transfers (submission/megha/db/schema.sql).

Table	Type	Notes
customer	strong	id, name, email (unique), timestamps
wallet	strong	belongs to customer, currency, status enum
transfer	strong	source/dest wallet FKs, amount (checked > 0), status enum, CHECK (source != dest), CHECK (settled_at >= created_at)
wallet_balance_history	weak (depends on wallet)	append-only balance snapshots
payment_attempt	weak (depends on transfer)	unique (transfer_id, attempt_no)

Every table carries updated_at and a soft-delete is_deleted flag — this is the contract the CDC layer depends on (see §2). A cdc_checkpoint table tracks per-table extraction cursors.

2. CDC Strategy

Implemented as polling-based CDC (submission/megha/cdc_extract.py), not WAL/Debezium — an explicit, documented simplification given the time budget. Each run:

1. Reads the last (updated_at, id) cursor per table from cdc_checkpoint.
2. Selects rows where (updated_at, id) > (last_updated_at, last_id) — a compound cursor, so multiple rows sharing an exact timestamp are never silently skipped.
3. Appends each changed row as a change record to lake/{table}.jsonl (op: upsert or op: delete, driven by is_deleted).
4. Advances the checkpoint only after the lake write succeeds, and commits both together — so a crash mid-run is safe to just re-run.

Verified idempotent: running the extractor twice with no source changes produces 0 new records both times (see test_extractor_is_idempotent_with_no_new_data).

Known limitation (documented, not hidden): because this is poll-based, a row that changes twice between two polls is only captured once, at its state at poll time — intermediate states are lost. A WAL-based approach (Debezium / logical replication) would capture every individual write. This was observed directly in testing: a wallet that was inserted and then soft-deleted within the same poll window shows up in the lake as a single delete record, never as an insert followed by a delete.

3. Lake / Warehouse Modeling

• Lake (lake/*.jsonl): append-only, one file per source table, one JSON record per captured change. This is the full-history source of truth and the basis for arbitrary-point replay.
• Warehouse (warehouse.* schema in the same Postgres instance, kept logically separate from source data via schema namespacing):
  • Snapshot tables (warehouse.customer, .wallet, .transfer, .wallet_balance_history, .payment_attempt) — one row per entity, latest state, upserted from the lake.
  • SCD2 history tables (warehouse.wallet_history, .transfer_history) — for the two entities where restore/time-travel matters most. valid_from / valid_to / is_current track state transitions; a row is only closed out and replaced if its tracked columns actually changed (avoids junk history entries from metadata-only touches).
  • warehouse.load_log makes the loader idempotent — verified by test_loader_is_idempotent.

4. Schema-Change Safety

submission/megha/schema_guard.py snapshots the live source schema (information_schema.columns) as a baseline on first run (--init). Every subsequent run diffs live vs. baseline:

• Safe changes (new column added) → logged, baseline updated, ingestion proceeds (exit 0).
• Breaking changes (column dropped, type changed, nullable tightened to NOT NULL, table dropped) → ingestion halts before any lake write, with the specific violation printed, exit code 1.

This is meant to gate cdc_extract.py in a real pipeline (e.g. CI/cron step: schema_guard.py && cdc_extract.py). Proven end-to-end in test_dropped_column_is_detected_as_breaking, which adds a real column, initializes a baseline, drops it, and asserts the halt + diagnostic message.

5. Time Travel / Restore

Recent state: query warehouse.wallet_history / warehouse.transfer_history for the row where valid_from <= T AND (valid_to > T OR valid_to IS NULL) — reconstructs state as of any timestamp T without touching the lake.

Older / full restore: replay lake/{table}.jsonl in order up to a timestamp into a scratch table — the lake retains complete history regardless of what the warehouse's SCD2 tables have compacted.

6. Validation / Parity

submission/megha/test_pipeline.py — 15 pytest tests across the 4 required categories, all passing:

• Modeling/constraints (5): positive amount, no self-transfer, settlement ordering, unique email, unique attempt number.
• CDC correctness (3): idempotent replay, lake record shape, delete rows correctly flagged op: delete.
• Warehouse correctness (5): snapshot row-count and content parity with source, SCD2 exactly-one-current-row invariant, closed SCD2 rows have valid_to set, idempotent loading.
• Schema-change safety (2): no false positives on an unchanged schema; a real dropped column is detected and halts ingestion.

Built Red → Blue → Green per requirement: each test was written to fail against the missing/naive behavior first, then made to pass with the minimal necessary implementation.

7. Catalog Exposure

submission/megha/catalog.yaml — lists all 14 datasets (5 lake, 7 warehouse, 2 operational) with location, format, owner, source table, update cadence, and grain. Validated as parseable YAML. In production this would be published to a real catalog (DataHub / Glue / Unity Catalog); this file captures the equivalent metadata manually, scoped to what this assignment produces.

Responsible AI Usage

I used an AI assistant (Claude) to help scaffold this solution — generating initial drafts of the schema DDL, CDC extractor, warehouse loader, schema-change guard, test suite, and this write-up, based on my own architecture decisions (domain choice, polling-CDC vs. WAL, SCD2-for-wallet/transfer-only, JSON-lines vs. Parquet). I ran every script and test myself, debugged real failures that came up during testing (e.g. a path-resolution bug in the test suite, a leftover column from an earlier failed test run), and verified the actual output at each step (row counts, SCD2 history contents, exit codes) rather than trusting generated code blindly. The reasoning behind each design decision above reflects my own understanding, checked against what I observed running the pipeline.