Assignment submission

submission/sample-candidate/approach_explanation.txt

@@ -0,0 +1,42 @@
+Pull Request: Technical Approach for CDC Pipeline
+What is this?
+Before writing the final code for the assignment, this dummy PR maps out the technical plan for building a reliable Change Data Capture (CDC) platform using Python and DuckDB, based on the criteria in the DATA_ASSIGNMENT.md.
+
+
+How it Works (The 4 Layers)
+1. The Source System (source_db.py)
+Instead of a single flat file, we use a proper relational setup (like customers and orders).
+
+Strict Rules: Financial data uses exact DECIMAL(18,2) types instead of unsafe floats, and statuses are restricted using database enums.
+
+2. The Ingestion Guard (cdc_pipeline.py)
+This moves data safely from the source into the lakehouse.
+
+Fail-Closed Schema Check: Before pulling data, the pipeline verifies the source structure against a contract. If a column is dropped or changed upstream, the pipeline stops immediately (SchemaMismatchException) so it doesn't corrupt anything downstream.
+
+Checkpoints: It tracks data using timestamps. If the pipeline crashes mid-run, it resumes exactly where it left off without duplicating data or creating gaps.
+
+3. The Data Lake: Absolute History (lake_append.py)
+The Data Lake is an append-only ledger. It never updates or overwrites records. Every single change (Insert, Update, Delete) creates a new row stamped with metadata:
+
+_cdc_op_type: Was it an (I)nsert, (U)pdate, or (D)elete?
+
+_cdc_extracted_at: When did we pull it?
+
+_cdc_lsn_version: The exact sequence order.
+
+4. The Warehouse: Clean Analytics & Time Travel (warehouse_snapshot.py)
+The Warehouse provides a clean view of the latest data for end users.
+
+The Latest State: It reads the lake's history and runs a window function (ROW_NUMBER() OVER) to grab only the latest update for each record. Deleted rows are filtered out automatically.
+
+Time Travel: Because the lake contains an unbroken history of changes, you can view exactly what the warehouse looked like at any point in the past just by filtering the query by a target timestamp (WHERE _cdc_extracted_at <= target_time).
+
+Testing the Edge Cases
+The test suite skips basic checks and focuses heavily on failure recovery:
+
+The Lifecycle Test: Inserts, updates, and deletes a record. It ensures the lake records all 3 events, while the warehouse correctly shows 0 active rows.
+
+The Circuit Breaker Test: Drops a source column to prove the pipeline actively stops running.
+
+The Time-Travel Test: Queries the warehouse at a specific historical timestamp to verify it rolls back accurately.

submission/sample-candidate/pipeline/cdc.py

@@ -1,89 +1,48 @@
-"""
-CDC capture layer.
-
-Simulates WAL-based change capture: every insert/update/delete on the source
-produces a CDCRecord with a monotonically increasing sequence number.
-
-Replay safety: callers can checkpoint the last processed sequence and call
-records_since(offset) to replay only unprocessed changes after a restart.
-"""
-
-from __future__ import annotations
-
-from dataclasses import dataclass, field
-from datetime import datetime, timezone
-from typing import Any
-
-VALID_OPERATIONS = frozenset({"insert", "update", "delete"})
-
-
-@dataclass
-class CDCRecord:
-    operation: str
-    table: str
-    primary_key: str
-    data: dict[str, Any]
-    captured_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
-    sequence: int = 0
-
-    def __post_init__(self) -> None:
-        if self.operation not in VALID_OPERATIONS:
-            raise ValueError(
-                f"Invalid CDC operation {self.operation!r}. "
-                f"Must be one of: {sorted(VALID_OPERATIONS)}"
-            )
-
-
-class CDCCapture:
-    """
-    In-process CDC log.
-
-    Production analogue: a Debezium/Kafka connector reading Postgres WAL.
-    Each record carries a sequence number equivalent to a Kafka offset or
-    Postgres LSN for checkpoint-based replay.
-    """
-
-    def __init__(self) -> None:
-        self._log: list[CDCRecord] = []
-        self._seq: int = 0
-
-    # ── public write API ─────────────────────────────────────────────────────
-
-    def insert(self, table: str, pk: str, data: dict[str, Any]) -> CDCRecord:
-        return self._record("insert", table, pk, data)
-
-    def update(self, table: str, pk: str, data: dict[str, Any]) -> CDCRecord:
-        return self._record("update", table, pk, data)
-
-    def delete(self, table: str, pk: str, data: dict[str, Any]) -> CDCRecord:
-        return self._record("delete", table, pk, data)
-
-    # ── public read / replay API ─────────────────────────────────────────────
-
-    def records_since(self, offset: int = 0) -> list[CDCRecord]:
-        """Return all records with sequence > offset (checkpoint replay)."""
-        return [r for r in self._log if r.sequence > offset]
-
-    @property
-    def latest_sequence(self) -> int:
-        return self._seq
-
-    @property
-    def log(self) -> list[CDCRecord]:
-        return list(self._log)
-
-    # ── internal ─────────────────────────────────────────────────────────────
-
-    def _record(
-        self, operation: str, table: str, pk: str, data: dict[str, Any]
-    ) -> CDCRecord:
-        self._seq += 1
-        rec = CDCRecord(
-            operation=operation,
-            table=table,
-            primary_key=pk,
-            data=data,
-            sequence=self._seq,
-        )
-        self._log.append(rec)
-        return rec
+import datetime
+from source.models import SCHEMA_CONTRACT
+
+class SchemaMismatchException(Exception): pass
+
+class CDCPipeline:
+    def __init__(self, source_conn, lake_conn):
+        self.source_conn = source_conn
+        self.lake_conn = lake_conn
+        self.high_watermarks = {table: datetime.datetime.min for table in SCHEMA_CONTRACT.keys()}
+
+    def verify_schema_contract(self):
+        """Active safety sentinel. Throws exception if upstream columns are corrupted."""
+        for table_name, expected_cols in SCHEMA_CONTRACT.items():
+            actual_cols = self.source_conn.execute(f"PRAGMA table_info('{table_name}')").fetchall()
+            actual_col_names = [col[1] for col in actual_cols]
+
+            for col in expected_cols:
+                if col not in actual_col_names:
+                    raise SchemaMismatchException(
+                        f"CRITICAL DRIFT DETECTED: Column '{col}' is missing on source table '{table_name}'!"
+                    )
+
+    def ingest_table_changes(self, table_name: str, lsn: int):
+        """Pulls incremental change logs past the last recorded high-watermark."""
+        self.verify_schema_contract()
+        now = datetime.datetime.now()
+
+        time_col = "created_at" if table_name == "transactions" else "updated_at"
+        cols_str = ", ".join(SCHEMA_CONTRACT[table_name])
+
+        changes = self.source_conn.execute(f"""
+            SELECT {cols_str} FROM {table_name} WHERE {time_col} > ? ORDER BY {time_col} ASC
+        """, (self.high_watermarks[table_name],)).fetchall()
+
+        if not changes:
+            return
+
+        for row in changes:
+            # Map mutations safely. If it's the initial pull, flag as Insert; otherwise Update
+            op_type = 'I' if self.high_watermarks[table_name] == datetime.datetime.min else 'U'
+            placeholders = ", ".join(["?"] * (len(row) + 3))
+
+            self.lake_conn.execute(f"""
+                INSERT INTO lake_{table_name}_history VALUES ({placeholders})
+            """, (*row, op_type, now, lsn))
+
+        self.high_watermarks[table_name] = max([row[-1] for row in changes])

submission/sample-candidate/pipeline/lake.py

@@ -1,69 +1,22 @@
-"""
-Lake layer — append-only storage for every CDC event.
-
-Every change is written exactly once. The lake is the source of truth for
-point-in-time replay and historical reconstruction.
-
-Production analogue: Parquet/Delta files on S3 or GCS, partitioned by
-table_name and captured_at date. No row is ever modified or deleted.
-"""
-
-from __future__ import annotations
-
-import json
-from datetime import datetime
-
 import duckdb
-
-from pipeline.cdc import CDCRecord
-
-
-def create_lake_table(conn: duckdb.DuckDBPyConnection) -> None:
-    conn.execute("""
-        CREATE TABLE IF NOT EXISTS lake_cdc_events (
-            sequence     INTEGER  NOT NULL,
-            operation    VARCHAR  NOT NULL,
-            table_name   VARCHAR  NOT NULL,
-            primary_key  VARCHAR  NOT NULL,
-            data         VARCHAR  NOT NULL,
-            captured_at  TIMESTAMP NOT NULL
-        )
-    """)
-
-
-def append_to_lake(conn: duckdb.DuckDBPyConnection, records: list[CDCRecord]) -> int:
-    """
-    Append CDC records to the lake.
-
-    Returns the number of records written.
-    Idempotency note: in production, deduplicate by sequence before appending.
-    """
-    if not records:
-        return 0
-
-    rows = [
-        (
-            r.sequence,
-            r.operation,
-            r.table,
-            r.primary_key,
-            _serialize(r.data),
-            r.captured_at,
-        )
-        for r in records
-    ]
-    conn.executemany(
-        "INSERT INTO lake_cdc_events VALUES (?, ?, ?, ?, ?, ?)",
-        rows,
-    )
-    return len(rows)
-
-
-def _serialize(data: dict) -> str:
-    return json.dumps(data, default=_json_default)
-
-
-def _json_default(obj: object) -> str:
-    if isinstance(obj, datetime):
-        return obj.isoformat()
-    raise TypeError(f"Object of type {type(obj).__name__} is not JSON serializable")
+from source.models import SCHEMA_CONTRACT
+
+def init_lake_store(source_conn: duckdb.DuckDBPyConnection) -> duckdb.DuckDBPyConnection:
+    """Creates the structural schema for an immutable append-only historical audit ledger."""
+    lake_conn = duckdb.connect(':memory:')
+
+    for table_name in SCHEMA_CONTRACT.keys():
+        source_info = source_conn.execute(f"PRAGMA table_info('{table_name}')").fetchall()
+        col_definitions = [f"{col[1]} {col[2]}" for col in source_info]
+
+        # Inject auditing metadata fields required to guarantee perfect recovery tracking
+        col_definitions.extend([
+            "_cdc_op_type CHAR(1)",
+            "_cdc_extracted_at TIMESTAMP",
+            "_cdc_lsn_version BIGINT"
+        ])
+
+        definitions_str = ", ".join(col_definitions)
+        lake_conn.execute(f"CREATE TABLE lake_{table_name}_history ({definitions_str});")
+
+    return lake_conn