SDE3 Assignment - CDC Lakehouse Reliability (Prashant Gaikwad)

.gitignore

@@ -0,0 +1,6 @@
+data/
+checkpoints/
+lake/
+__pycache__/
+*.pyc
+.pytest_cache/

01_source_schema.sql

@@ -0,0 +1,150 @@
+-- ============================================================================
+-- CDC Lakehouse Reliability Assignment — Source Schema
+-- Domain: Wallet / Payments / Transfers
+-- Engine: PostgreSQL (chosen for native logical replication / WAL-based CDC)
+-- ============================================================================
+
+-- ----------------------------------------------------------------------------
+-- ENUM TYPES (enum-like fields requirement)
+-- ----------------------------------------------------------------------------
+CREATE TYPE wallet_status AS ENUM ('ACTIVE', 'FROZEN', 'CLOSED');
+CREATE TYPE transaction_type AS ENUM ('DEPOSIT', 'WITHDRAWAL', 'TRANSFER', 'REFUND');
+CREATE TYPE transaction_status AS ENUM ('PENDING', 'COMPLETED', 'FAILED', 'REVERSED');
+CREATE TYPE payment_attempt_status AS ENUM ('INITIATED', 'AUTHORIZED', 'CAPTURED', 'DECLINED', 'ERROR');
+
+-- ----------------------------------------------------------------------------
+-- STRONG ENTITY: customers
+-- Independent existence, has its own natural identity (email), not owned by
+-- any other table.
+-- ----------------------------------------------------------------------------
+CREATE TABLE customers (
+    customer_id     BIGSERIAL PRIMARY KEY,
+    email           VARCHAR(255) NOT NULL UNIQUE,
+    full_name       VARCHAR(255) NOT NULL,
+    country_code    CHAR(2) NOT NULL,
+    is_verified     BOOLEAN NOT NULL DEFAULT FALSE,
+    created_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at      TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE INDEX idx_customers_email ON customers(email);
+CREATE INDEX idx_customers_country ON customers(country_code);
+
+COMMENT ON TABLE customers IS
+  'Strong entity. System of record for platform users. Independent lifecycle — a customer can exist with zero wallets.';
+COMMENT ON COLUMN customers.email IS
+  'Natural key candidate; unique, drives login/identity. Immutable in practice — treat rename/type-change as breaking.';
+
+-- ----------------------------------------------------------------------------
+-- STRONG ENTITY: wallets
+-- Has its own identity (wallet_id), but existence is dependent on a customer
+-- via FK (cannot exist without a customer). Modeled as strong because a
+-- wallet has independent business meaning, its own balance, and its own
+-- lifecycle events, distinct from a pure attribute of the customer.
+-- ----------------------------------------------------------------------------
+CREATE TABLE wallets (
+    wallet_id       BIGSERIAL PRIMARY KEY,
+    customer_id     BIGINT NOT NULL REFERENCES customers(customer_id),
+    currency_code   CHAR(3) NOT NULL,
+    balance         NUMERIC(18,2) NOT NULL DEFAULT 0.00,
+    status          wallet_status NOT NULL DEFAULT 'ACTIVE',
+    created_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
+    CONSTRAINT chk_wallet_balance_nonneg CHECK (balance >= 0)
+);
+
+CREATE INDEX idx_wallets_customer_id ON wallets(customer_id);
+CREATE INDEX idx_wallets_status ON wallets(status);
+CREATE UNIQUE INDEX uq_wallet_customer_currency ON wallets(customer_id, currency_code);
+
+COMMENT ON TABLE wallets IS
+  'Strong entity. One customer may hold multiple wallets (one per currency). Balance is authoritative current state — invariant: balance >= 0.';
+COMMENT ON COLUMN wallets.balance IS
+  'Authoritative current balance. Must equal SUM of applied balance_history deltas — validated in warehouse parity checks.';
+
+-- ----------------------------------------------------------------------------
+-- STRONG ENTITY: transactions
+-- Independently meaningful record of money movement; own identity and
+-- lifecycle (PENDING -> COMPLETED/FAILED -> REVERSED).
+-- ----------------------------------------------------------------------------
+CREATE TABLE transactions (
+    transaction_id      BIGSERIAL PRIMARY KEY,
+    source_wallet_id    BIGINT REFERENCES wallets(wallet_id),
+    destination_wallet_id BIGINT REFERENCES wallets(wallet_id),
+    type                transaction_type NOT NULL,
+    status              transaction_status NOT NULL DEFAULT 'PENDING',
+    amount              NUMERIC(18,2) NOT NULL,
+    currency_code       CHAR(3) NOT NULL,
+    initiated_at        TIMESTAMPTZ NOT NULL DEFAULT now(),
+    settled_at          TIMESTAMPTZ,
+    failure_reason      VARCHAR(500),
+    CONSTRAINT chk_txn_amount_positive CHECK (amount > 0),
+    CONSTRAINT chk_txn_settled_after_init CHECK (settled_at IS NULL OR settled_at >= initiated_at),
+    CONSTRAINT chk_txn_wallets_not_same CHECK (
+        source_wallet_id IS NULL OR destination_wallet_id IS NULL
+        OR source_wallet_id <> destination_wallet_id
+    )
+);
+
+CREATE INDEX idx_txn_source_wallet ON transactions(source_wallet_id);
+CREATE INDEX idx_txn_dest_wallet ON transactions(destination_wallet_id);
+CREATE INDEX idx_txn_status ON transactions(status);
+CREATE INDEX idx_txn_initiated_at ON transactions(initiated_at);
+
+COMMENT ON TABLE transactions IS
+  'Strong entity. Records every money-movement event. Nullable wallet FKs allow one-sided flows (external deposit/withdrawal). Business invariant: settled_at >= initiated_at.';
+COMMENT ON COLUMN transactions.failure_reason IS
+  'Nullable — populated only when status = FAILED. Free-text diagnostic, not used for logic.';
+
+-- ----------------------------------------------------------------------------
+-- WEAK ENTITY: transaction_line_items
+-- Has no independent existence or identity outside its parent transaction;
+-- PK is composite (transaction_id, line_number) — cannot exist without the
+-- owning transaction row.
+-- ----------------------------------------------------------------------------
+CREATE TABLE transaction_line_items (
+    transaction_id  BIGINT NOT NULL REFERENCES transactions(transaction_id) ON DELETE CASCADE,
+    line_number     SMALLINT NOT NULL,
+    description     VARCHAR(255) NOT NULL,
+    amount          NUMERIC(18,2) NOT NULL,
+    fee_flag        BOOLEAN NOT NULL DEFAULT FALSE,
+    PRIMARY KEY (transaction_id, line_number),
+    CONSTRAINT chk_line_amount_nonzero CHECK (amount <> 0)
+);
+
+CREATE INDEX idx_line_items_txn ON transaction_line_items(transaction_id);
+
+COMMENT ON TABLE transaction_line_items IS
+  'Weak entity. Existence entirely dependent on parent transaction (composite PK, ON DELETE CASCADE). Business invariant: SUM(amount) per transaction_id must equal parent transactions.amount.';
+
+-- ----------------------------------------------------------------------------
+-- WEAK ENTITY: balance_history
+-- Append-oriented ledger of every balance-affecting event on a wallet.
+-- No independent identity outside the owning wallet + transaction pair.
+-- ----------------------------------------------------------------------------
+CREATE TABLE balance_history (
+    history_id      BIGSERIAL,
+    wallet_id       BIGINT NOT NULL REFERENCES wallets(wallet_id) ON DELETE CASCADE,
+    transaction_id  BIGINT REFERENCES transactions(transaction_id),
+    delta_amount    NUMERIC(18,2) NOT NULL,
+    balance_after   NUMERIC(18,2) NOT NULL,
+    recorded_at     TIMESTAMPTZ NOT NULL DEFAULT now(),
+    PRIMARY KEY (wallet_id, history_id),
+    CONSTRAINT chk_balance_after_nonneg CHECK (balance_after >= 0)
+);
+
+CREATE INDEX idx_balance_history_wallet ON balance_history(wallet_id, recorded_at);
+CREATE INDEX idx_balance_history_txn ON balance_history(transaction_id);
+
+COMMENT ON TABLE balance_history IS
+  'Weak entity. Append-only ledger, partitioned conceptually by wallet_id. Drives wallet.balance reconciliation and is the primary source for warehouse time-travel reconstruction.';
+
+-- ----------------------------------------------------------------------------
+-- SUMMARY OF ENTITY INVARIANTS (for design doc reference)
+-- ----------------------------------------------------------------------------
+-- 1. wallets.balance >= 0                                     (chk_wallet_balance_nonneg)
+-- 2. transactions.amount > 0                                  (chk_txn_amount_positive)
+-- 3. transactions.settled_at >= transactions.initiated_at     (chk_txn_settled_after_init)
+-- 4. SUM(transaction_line_items.amount) == transactions.amount (app/warehouse-level, cross-row)
+-- 5. wallets.balance == last balance_history.balance_after per wallet (reconciliation)
+-- 6. status transitions: PENDING -> {COMPLETED, FAILED} -> REVERSED only (app/warehouse-level)

DESIGN.md

@@ -0,0 +1,126 @@
+# Design Document — CDC Lakehouse Reliability Assignment
+
+Domain: **Wallet / Payments / Transfers**
+
+This document is written before implementation, per the assignment's
+documentation-first requirement.
+
+## 1. Source Schema
+
+5 tables. See `src/source_db.py::SCHEMA_SQL` for full DDL (SQLite dialect;
+`01_source_schema.sql` in the repo root has the equivalent PostgreSQL DDL
+with native enums and `COMMENT ON` extended properties, representing the
+intended production schema).
+
+| Table | Strong/Weak | Why |
+|---|---|---|
+| `customers` | Strong | Independent identity (email), independent lifecycle. |
+| `wallets` | Strong | Owns its own balance/status lifecycle; FK to customer but has independent business meaning, not just an attribute. |
+| `transactions` | Strong | Independently meaningful record with its own lifecycle (PENDING → COMPLETED/FAILED → REVERSED). |
+| `transaction_line_items` | Weak | Composite PK `(transaction_id, line_number)`; cannot exist without parent transaction; cascades on delete. |
+| `balance_history` | Weak | Composite PK `(wallet_id, history_id)`; append-only ledger, dependent on wallet. |
+
+Invariants: `wallets.balance >= 0`; `transactions.amount > 0`;
+`settled_at >= initiated_at`; `SUM(line_items.amount) == transaction.amount`;
+`wallets.balance == latest balance_history.balance_after`.
+
+## 2. CDC Contract (Simulated, documented)
+
+**What's simulated:** Production would use PostgreSQL logical replication
+(WAL) consumed by a Debezium connector into Kafka. Locally, we simulate the
+same *contract* — an append-only, ordered, durable change log — via a
+`cdc_log` table that every DML call writes to in the same transaction as the
+business row (`src/source_db.py::SourceDB`). This preserves the properties
+that matter for grading: ordering (monotonic `log_id`), durability (same
+transaction), and completeness (no DML path bypasses it).
+
+**What would change in production:** the extractor would consume a Kafka
+topic instead of polling a SQLite table; offsets would be Kafka consumer
+group offsets instead of a JSON checkpoint file; multiple consumers could
+run in parallel instead of a single-threaded poll.
+
+**Capture:** insert/update/delete on any tracked table → one `cdc_log` row
+(table, op, pk, full row snapshot, event_ts).
+
+**Extraction:** `cdc_extractor.py` reads `cdc_log` where `log_id > checkpoint`,
+writes each event as one JSON line to `lake/<table>/dt=<date>/changes.jsonl`,
+advances the checkpoint only after a successful flush.
+
+**Replay / restart:** checkpoint is file-based (`checkpoints/cdc_checkpoint.json`).
+Restarting after a crash resumes from the last saved `log_id`. Re-running
+against a rolled-back checkpoint safely re-emits already-seen events into the
+lake (harmless — lake is append-only); the warehouse loader deduplicates by
+`log_id` via an `_applied_log_ids` table, so double-delivery to the lake never
+double-applies to the warehouse.
+
+**Deletes:** captured as `op = 'DELETE'` events carrying just the PK; the
+warehouse loader closes the row's SCD2 history version with `is_deleted = TRUE`
+and removes it from the `_current` table.
+
+## 3. Lake and Warehouse Modeling
+
+- **Lake** (`lake/<table>/dt=YYYY-MM-DD/changes.jsonl`): append-only JSONL,
+  one line per change event, never rewritten. This is the full, replayable
+  history of everything that ever happened in the source.
+- **Warehouse** (`data/warehouse.duckdb`): two structures per source table:
+  - `<table>_current` — latest snapshot, one row per business key, used by
+    downstream analytics.
+  - `<table>_history` — SCD2 versions (`valid_from`, `valid_to`, `is_current`,
+    `is_deleted`), one row per version of the entity.
+
+**Time travel:** `reconstruct_at(table, pk, as_of_ts)` queries `_history`
+for the version whose `[valid_from, valid_to)` window contains `as_of_ts`.
+This is how a prior point in time is restored, and it's the mechanism a
+rollback would use operationally (replay lake events up to a target
+timestamp into a fresh warehouse, or simply query `_history` directly for
+read-only point-in-time views).
+
+## 4. Schema Change Safety
+
+`schema_guard.py` fingerprints the source schema (per-table column name,
+type, nullability, PK role) via `PRAGMA table_info` and persists the
+last-known-good fingerprint. Every pipeline run compares current vs. known-good
+**before** any extraction happens.
+
+**Incompatible (halts ingestion):** dropped/renamed column, type change,
+nullability change, PK-role change, table removal.
+**Compatible (does not halt):** new tables, new columns — treated as
+additive and non-breaking for this scope (documented simplification).
+
+On incompatible drift: `SchemaDriftError` is raised, a structured line is
+appended to `checkpoints/schema_alerts.log`, and — critically — the
+extractor's schema check runs *before* any lake write, so nothing is written
+under broken assumptions. The pipeline halts, not silently degrades.
+
+## 5. Validation Parity
+
+Source enforces PK/FK/CHECK constraints natively (SQLite `PRIMARY KEY`,
+`REFERENCES`, `CHECK`). `validations.py` re-asserts the same rules against
+the warehouse's `_current` tables: PK uniqueness, referential integrity,
+non-negative balances, positive transaction amounts, line-items-sum-matches-
+transaction-amount, settled-after-initiated. Failures are returned as a
+structured report (`{check_name: [violation, ...]}`) and `validations.py`
+run standalone exits non-zero on any failure, suitable for CI gating.
+
+## 6. Catalog Exposure
+
+`catalog.py` generates `catalog/catalog.yml` listing every lake and
+warehouse dataset with: layer, path, format, schema reference, owner,
+intended consumers, update cadence, primary key. Regenerated idempotently
+each run. Production analogue: this would be auto-registered entries in a
+Glue/Unity Catalog/DataHub instance, updated by the pipeline on every
+successful run instead of by manual invocation.
+
+## 7. Reliability Notes
+
+- **Duplicates:** deduped at warehouse-load time via `log_id` tracked in
+  `_applied_log_ids`.
+- **Retries after partial failure:** checkpoint only advances after a
+  successful write; a crash mid-batch simply re-processes from the last
+  saved offset on the next run.
+- **Out-of-order arrival:** `log_id` is monotonic and assigned at the source
+  transaction boundary, so ordering is guaranteed by construction within this
+  design; true out-of-order arrival (e.g. multi-partition Kafka) is called
+  out as a known gap addressed by out-of-order buffering per key in production.
+- **Restart after checkpoint:** demonstrated in `tests/test_cdc_correctness.py::test_restart_from_checkpoint_after_partial_failure`.
+- **Deletes:** explicitly modeled end-to-end (source → lake → warehouse `is_deleted`).