Submission of Infra assessment | Implement reliable CDC pipeline, lakehouse storage, schema contract safety, and catalog metadata

[ashokchetan800]

Summary

This pull request delivers a complete, production-grade Change Data Capture (CDC) Lakehouse Reliability implementation for a Payments & Digital Wallet system using DuckDB and Python.

The architecture guarantees:

• Full append-only change history in the Data Lake (Bronze layer).
• Near real-time current snapshots and point-in-time time travel in the Data Warehouse (Silver layer).
• Strict schema contract enforcement ("stop-the-line" safe halting on incompatible schema drift).
• Validation parity between source constraints and downstream warehouse models.
• Published dataset catalog metadata and an automated suite of unit/integration tests and CI checks.

---

Source Schema Design

• Domain chosen: Digital Payments & Wallet Ledger System (5 relational tables).
• Strong entities:
  • customers: Account profile container (customer_id PK, name, email, status enum).
  • wallets: Financial balance container (wallet_id PK, customer_id FK, balance DECIMAL(18,2) $\ge 0$, currency, status enum).
  • merchants: Registered merchant container (merchant_id PK, name, email, status enum).
• Weak entities:
  • transactions: Payment lifecycle event (transaction_id PK, wallet_id FK, merchant_id FK, amount $> 0$, fee $\ge 0$, direction enum, status enum, settled_at $\ge$ created_at).
  • ledger_entries: Double-entry accounting audit record (entry_id PK, transaction_id FK, wallet_id FK, entry_type enum, amount $> 0$, balance_after $\ge 0$).
• Keys, relationships, and indexes: Primary keys on all tables, foreign keys linking dependent entities, and indexes on customer_id, wallet_id, status, and transaction_id.
• Source validation rules: Enforced DDL CHECK constraints (non-negative balances/fees, positive transaction amounts, valid status enums, timestamp ordering).

---

CDC Strategy

• How changes are captured: Monotonically increasing sequence allocation (1, 2, 3...) tracking insert, update, and delete operations via CDCCapture.
• How inserts, updates, and deletes are handled: Every operation is logged to the Lake. In the Warehouse, insert/update upserts entity state, while delete marks soft-delete flag _deleted = true.
• How replay/restart works: Checkpointing via records_since(offset), allowing pipeline consumers to resume cleanly after network partitions or service restarts without missing events.
• How duplicates are handled: Sequence-level deduplication before appending to the Lake (append_to_lake), and primary key upsert idempotency in the Warehouse (apply_cdc_records).

---

Lake and Warehouse Modeling

• How the lake captures every change: Append-only lake_cdc_events table storing un-mutated raw CDC event payloads (sequence, operation, table_name, primary_key, data, captured_at).
• How the warehouse maintains latest snapshot: Silver layer tables (wh_*) reflecting the latest operational snapshot keyed on entity primary keys with _cdc_seq and _deleted flags.
• How time travel / restore is supported: Engine function reconstruct_snapshot_at(conn, target_seq) replays Lake events up to sequence target_seq to reconstruct exact prior warehouse state at any historical moment.

---

Schema Change Safety

• How incompatible changes are detected: check_schema_compatibility() compares source DDL against SCHEMA_CONTRACT.
• How ingestion is stopped: If required columns are dropped, renamed, or missing, assert_schema_compatible() logs a STOP-THE-LINE alert and raises SchemaIncompatibilityError.
• How warnings/failures are surfaced: Emits structured console/log warnings and exits CI scripts with non-zero exit code 1.

---

Validation Parity

• Which source system validations were mirrored downstream: Primary key uniqueness, foreign key referential integrity, NOT NULL rules, non-negative wallet balances, positive transaction amounts, timestamp ordering (settled_at >= created_at), and status enum compliance.
• How failures are checked and reported: Automated validation suite in pipeline/validations.py executed by scripts/run_data_quality_checks.py returning exit code 0 on success or 1 on failure.

---

Catalog Exposure

Exposed metadata for Lake (lake_cdc_events) and Warehouse (wh_*) datasets in catalog/catalog.json. Metadata includes layer classification (lake vs warehouse), owner teams, update cadences, intended audiences, and column-level schemas with data types and descriptions. Validated via scripts/validate_catalog.py.

---

Validation

Ran the complete test suite and CI verification scripts locally:

1. pytest tests/ (18/18 passed)
2. python scripts/check_schema_contracts.py (Exit 0 — all 5 tables passed)
3. python scripts/run_data_quality_checks.py (Exit 0 — all system & business rules passed)
4. python scripts/validate_catalog.py (Exit 0 — catalog metadata valid)
5. ruff check . & black --check . (Passed with zero errors)

---

Known Limitations / Next Steps

• Message Transport: Currently simulated via in-process memory stream CDCCapture. In production, this would use Debezium reading Postgres WAL into Apache Kafka.
• Storage Format: Currently DuckDB tables. In production, Lake tables would be stored as Apache Iceberg / Delta Lake Parquet files on AWS S3.
• Time Travel Optimization: Point-in-time recovery currently replays Lake events. In production, building SCD Type 2 dimension tables via dbt snapshot models would provide instant SQL time-travel queries without full event replay.

---

Responsible AI Usage

• Did you use AI tools?: Yes, utilized AI coding assistance (Antigravity by Google DeepMind) for architectural planning, code scaffolding, test suite design, and documentation.
• Where did they help?: Accelerated boilerplating of DuckDB DDLs, pytest fixtures, ruff/black formatting alignment, and validation parity assertions.
• What did you personally verify or correct?: Reviewed, tested, and validated all SQL queries, DuckDB constraint execution behavior, sequence deduplication logic, time-travel reconstruction math, and schema contract failure triggers.

---

Author Checklist

• Linting passes (ruff & black)
• Tests pass (pytest 18/18 passed)
• Model validation/tests pass
• Schema compatibility checks pass (scripts/check_schema_contracts.py)
• Data quality validations pass (scripts/run_data_quality_checks.py)
• Catalog metadata validation passes (scripts/validate_catalog.py)
• README was tested from a clean setup
• End-to-end CDC flow was validated locally