Implement data-assignment for ecommerce pipeline

submission/cdc-lakehouse/.gitignore

@@ -0,0 +1,8 @@
+lake/
+warehouse/
+state/
+ingest/FAILED_SCHEMA_CHANGE.md
+__pycache__/
+*.pyc
+.pytest_cache/
+.venv/

submission/cdc-lakehouse/README.md

@@ -0,0 +1,88 @@
+# CDC Lakehouse — E-commerce (Interview MVP)
+
+Minimal CDC pipeline: **MySQL** source (trigger outbox) → **Parquet lake** (append-only events) → **DuckDB warehouse** (current-state snapshots) with schema contract gate, data quality checks, catalog, and point-in-time restore from the lake.
+
+## Architecture
+
+```
+MySQL business tables
+  └─ triggers → cdc_events (I/U/D)
+        └─ Python ingest (checkpoint on event_id)
+              ├─ schema fingerprint + contract check (stop-the-line)
+              ├─ lake/source_table=*/dt=*/*.parquet
+              └─ warehouse/warehouse.duckdb (wh_* tables)
+```
+
+**Production analogue:** Debezium/binlog → Kafka → object storage (Iceberg/Delta) → Snowflake/BigQuery; schema registry for contracts.
+
+## Prerequisites
+
+- Docker (for MySQL)
+- Python 3.11+
+
+## Quick start
+
+```bash
+cd submission/cdc-lakehouse
+python3 -m venv .venv && source .venv/bin/activate
+pip install -r requirements.txt
+
+docker compose up -d
+# Wait for MySQL healthy
+
+python -m pipeline.ingest
+python scripts/run_data_quality_checks.py
+python scripts/validate_catalog.py
+pytest -q
+```
+
+## Restore (time travel)
+
+Rebuild warehouse state from lake events up to a timestamp:
+
+```bash
+python scripts/restore_warehouse.py --as-of 2026-05-26T12:00:00+00:00
+```
+
+## Project layout
+
+| Path | Purpose |
+|------|---------|
+| `source/schema.sql` | MySQL DDL (6 tables + `cdc_events`) |
+| `source/triggers.sql` | CDC outbox triggers |
+| `source/seed.sql` | Sample transactional data |
+| `pipeline/ingest.py` | Ingestion entrypoint |
+| `contracts/` | Schema contract + fingerprint baseline |
+| `lake/` | Append-only Parquet CDC events (generated) |
+| `warehouse/warehouse.duckdb` | Current-state analytics tables (generated) |
+| `catalog/catalog.json` | Dataset metadata |
+| `scripts/` | Quality, catalog, restore, contract checks |
+
+## Schema change safety (source-owned contract)
+
+The **source team** owns the schema. Data engineering does not bootstrap truth from live MySQL.
+
+| Artifact | Owner | Purpose |
+|----------|-------|---------|
+| `contracts/schema_contract_spec.json` | Source team | Authoritative columns, types, nullability, keys |
+| `contracts/schema_fingerprint.json` | Compiled from contract | Expected structural hash per table |
+
+**Every ingestion** validates live `INFORMATION_SCHEMA` against the contract (columns, types, unexpected columns, fingerprint). On mismatch → **stop**, write `ingest/FAILED_SCHEMA_CHANGE.md`, exit non-zero.
+
+When the source team publishes a new schema version, update `schema_contract_spec.json` and recompile:
+
+```bash
+python scripts/compile_schema_fingerprint.py
+```
+
+## Tests
+
+```bash
+pytest -q
+```
+
+Unit tests cover lake I/O, warehouse merge/idempotency, contracts, restore filtering, and catalog shape. MySQL integration is exercised manually via Docker + `pipeline.ingest`.
+
+## Responsible AI
+
+Document in your PR how AI was used and what you validated manually.

submission/cdc-lakehouse/catalog/catalog.json

@@ -0,0 +1,67 @@
+{
+  "datasets": [
+    {
+      "name": "lake_cdc_events",
+      "layer": "lake",
+      "description": "Append-only Parquet CDC events partitioned by source_table and date. Full change history for audit and point-in-time replay.",
+      "owner": "data-platform",
+      "consumers": ["data-platform", "analytics", "audit"],
+      "update_cadence": "near-real-time",
+      "location": "lake/source_table=*/dt=*/*.parquet",
+      "schema": {
+        "event_id": "BIGINT — monotonic CDC offset",
+        "source_table": "VARCHAR — business table name",
+        "op": "CHAR(1) — I | U | D",
+        "pk_json": "JSON — primary key",
+        "row_json": "JSON — row snapshot",
+        "event_ts": "TIMESTAMP — capture time",
+        "ingested_at": "TIMESTAMP — pipeline write time"
+      }
+    },
+    {
+      "name": "wh_customers",
+      "layer": "warehouse",
+      "description": "Current-state customer snapshot from CDC merge.",
+      "owner": "data-platform",
+      "consumers": ["analytics", "product"],
+      "update_cadence": "near-real-time",
+      "location": "warehouse/warehouse.duckdb#wh_customers",
+      "schema": {
+        "customer_id": "BIGINT PK",
+        "_cdc_event_id": "BIGINT — last applied event",
+        "_deleted": "BOOLEAN — tombstone flag"
+      }
+    },
+    {
+      "name": "wh_orders",
+      "layer": "warehouse",
+      "description": "Current-state orders for analytics; supports restore via lake replay.",
+      "owner": "data-platform",
+      "consumers": ["analytics", "finance"],
+      "update_cadence": "near-real-time",
+      "location": "warehouse/warehouse.duckdb#wh_orders",
+      "schema": {
+        "order_id": "BIGINT PK",
+        "order_status": "ENUM-like VARCHAR",
+        "order_total": "DECIMAL(12,2)",
+        "_cdc_event_id": "BIGINT",
+        "_deleted": "BOOLEAN"
+      }
+    },
+    {
+      "name": "wh_order_items",
+      "layer": "warehouse",
+      "description": "Current-state order line items.",
+      "owner": "data-platform",
+      "consumers": ["analytics"],
+      "update_cadence": "near-real-time",
+      "location": "warehouse/warehouse.duckdb#wh_order_items",
+      "schema": {
+        "order_item_id": "BIGINT PK",
+        "line_total": "DECIMAL(12,2)",
+        "_cdc_event_id": "BIGINT",
+        "_deleted": "BOOLEAN"
+      }
+    }
+  ]
+}

submission/cdc-lakehouse/conftest.py

@@ -0,0 +1,6 @@
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parent
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))

submission/cdc-lakehouse/contracts/schema_contract.json

@@ -0,0 +1,58 @@
+{
+  "customers": [
+    "customer_id",
+    "name",
+    "email",
+    "phone",
+    "status",
+    "created_at",
+    "updated_at",
+    "is_deleted"
+  ],
+  "order_items": [
+    "order_item_id",
+    "order_id",
+    "product_id",
+    "quantity",
+    "unit_price",
+    "line_total",
+    "is_deleted"
+  ],
+  "order_status_events": [
+    "status_event_id",
+    "order_id",
+    "from_status",
+    "to_status",
+    "event_at",
+    "is_deleted"
+  ],
+  "orders": [
+    "order_id",
+    "customer_id",
+    "order_status",
+    "order_total",
+    "created_at",
+    "updated_at",
+    "paid_at",
+    "cancel_reason",
+    "is_deleted"
+  ],
+  "payments": [
+    "payment_id",
+    "order_id",
+    "payment_amount",
+    "payment_status",
+    "created_at",
+    "updated_at",
+    "is_deleted"
+  ],
+  "products": [
+    "product_id",
+    "sku",
+    "name",
+    "unit_price",
+    "created_at",
+    "updated_at",
+    "is_deleted"
+  ]
+}

submission/cdc-lakehouse/contracts/schema_contract_spec.json

@@ -0,0 +1,75 @@
+{
+  "version": "1.0.0",
+  "published_by": "source-team",
+  "description": "Authoritative source schema contract. Updated only when the source team publishes a new schema version.",
+  "tables": {
+    "customers": {
+      "columns": [
+        {"name": "customer_id", "data_type": "bigint", "nullable": false, "column_key": "PRI"},
+        {"name": "name", "data_type": "varchar", "nullable": false, "column_key": ""},
+        {"name": "email", "data_type": "varchar", "nullable": true, "column_key": ""},
+        {"name": "phone", "data_type": "varchar", "nullable": true, "column_key": ""},
+        {"name": "status", "data_type": "enum", "nullable": false, "column_key": ""},
+        {"name": "created_at", "data_type": "timestamp", "nullable": false, "column_key": ""},
+        {"name": "updated_at", "data_type": "timestamp", "nullable": false, "column_key": ""},
+        {"name": "is_deleted", "data_type": "tinyint", "nullable": false, "column_key": ""}
+      ]
+    },
+    "products": {
+      "columns": [
+        {"name": "product_id", "data_type": "bigint", "nullable": false, "column_key": "PRI"},
+        {"name": "sku", "data_type": "varchar", "nullable": false, "column_key": "UNI"},
+        {"name": "name", "data_type": "varchar", "nullable": false, "column_key": ""},
+        {"name": "unit_price", "data_type": "decimal", "nullable": false, "column_key": ""},
+        {"name": "created_at", "data_type": "timestamp", "nullable": false, "column_key": ""},
+        {"name": "updated_at", "data_type": "timestamp", "nullable": false, "column_key": ""},
+        {"name": "is_deleted", "data_type": "tinyint", "nullable": false, "column_key": ""}
+      ]
+    },
+    "orders": {
+      "columns": [
+        {"name": "order_id", "data_type": "bigint", "nullable": false, "column_key": "PRI"},
+        {"name": "customer_id", "data_type": "bigint", "nullable": false, "column_key": "MUL"},
+        {"name": "order_status", "data_type": "enum", "nullable": false, "column_key": ""},
+        {"name": "order_total", "data_type": "decimal", "nullable": false, "column_key": ""},
+        {"name": "created_at", "data_type": "timestamp", "nullable": false, "column_key": ""},
+        {"name": "updated_at", "data_type": "timestamp", "nullable": false, "column_key": ""},
+        {"name": "paid_at", "data_type": "timestamp", "nullable": true, "column_key": ""},
+        {"name": "cancel_reason", "data_type": "varchar", "nullable": true, "column_key": ""},
+        {"name": "is_deleted", "data_type": "tinyint", "nullable": false, "column_key": ""}
+      ]
+    },
+    "order_items": {
+      "columns": [
+        {"name": "order_item_id", "data_type": "bigint", "nullable": false, "column_key": "PRI"},
+        {"name": "order_id", "data_type": "bigint", "nullable": false, "column_key": "MUL"},
+        {"name": "product_id", "data_type": "bigint", "nullable": false, "column_key": "MUL"},
+        {"name": "quantity", "data_type": "int", "nullable": false, "column_key": ""},
+        {"name": "unit_price", "data_type": "decimal", "nullable": false, "column_key": ""},
+        {"name": "line_total", "data_type": "decimal", "nullable": false, "column_key": ""},
+        {"name": "is_deleted", "data_type": "tinyint", "nullable": false, "column_key": ""}
+      ]
+    },
+    "payments": {
+      "columns": [
+        {"name": "payment_id", "data_type": "bigint", "nullable": false, "column_key": "PRI"},
+        {"name": "order_id", "data_type": "bigint", "nullable": false, "column_key": "MUL"},
+        {"name": "payment_amount", "data_type": "decimal", "nullable": false, "column_key": ""},
+        {"name": "payment_status", "data_type": "enum", "nullable": false, "column_key": ""},
+        {"name": "created_at", "data_type": "timestamp", "nullable": false, "column_key": ""},
+        {"name": "updated_at", "data_type": "timestamp", "nullable": false, "column_key": ""},
+        {"name": "is_deleted", "data_type": "tinyint", "nullable": false, "column_key": ""}
+      ]
+    },
+    "order_status_events": {
+      "columns": [
+        {"name": "status_event_id", "data_type": "bigint", "nullable": false, "column_key": "PRI"},
+        {"name": "order_id", "data_type": "bigint", "nullable": false, "column_key": "MUL"},
+        {"name": "from_status", "data_type": "enum", "nullable": true, "column_key": ""},
+        {"name": "to_status", "data_type": "enum", "nullable": false, "column_key": ""},
+        {"name": "event_at", "data_type": "timestamp", "nullable": false, "column_key": ""},
+        {"name": "is_deleted", "data_type": "tinyint", "nullable": false, "column_key": ""}
+      ]
+    }
+  }
+}

submission/cdc-lakehouse/contracts/schema_fingerprint.json

@@ -0,0 +1,8 @@
+{
+  "customers": "95fa05b2836a3633371d00722e29d11325c9801c23757165b97c55fb626ecf2e",
+  "order_items": "4882b46acd323bc65038eba095a0711a7ed8474c4f9d796b5e156554b74230e7",
+  "order_status_events": "4e0e33610345950cf5c80e202916fbdee9a9fa08566eacb9c80f1bb8528375dd",
+  "orders": "1a7cecbb5073d6bdf33fd90c0a898fd403b6c32883665279e099e1cec667306e",
+  "payments": "4572c2fe51c3e99a3ffc7106278bcea7db91e504c6c26a8055867f47ab4e1838",
+  "products": "74fac4af1e7164500ac39c9499dd86176f8ed59560f674f8fe845ef8d2a9966d"
+}

submission/cdc-lakehouse/docker-compose.yml

@@ -0,0 +1,21 @@
+services:
+  mysql:
+    image: mysql:8.0
+    container_name: cdc-lakehouse-mysql
+    command: --default-authentication-plugin=mysql_native_password
+    environment:
+      MYSQL_ROOT_PASSWORD: root
+      MYSQL_DATABASE: cdc_source
+      MYSQL_USER: cdc_user
+      MYSQL_PASSWORD: cdc_pass
+    ports:
+      - "3306:3306"
+    volumes:
+      - ./source/schema.sql:/docker-entrypoint-initdb.d/01-schema.sql:ro
+      - ./source/triggers.sql:/docker-entrypoint-initdb.d/02-triggers.sql:ro
+      - ./source/seed.sql:/docker-entrypoint-initdb.d/03-seed.sql:ro
+    healthcheck:
+      test: ["CMD", "mysqladmin", "ping", "-h", "localhost", "-uroot", "-proot"]
+      interval: 5s
+      timeout: 5s
+      retries: 10

submission/cdc-lakehouse/pipeline/__init__.py

@@ -0,0 +1 @@
+"""CDC lakehouse pipeline package."""

submission/cdc-lakehouse/pipeline/cdc_event.py

@@ -0,0 +1,53 @@
+"""CDC event model shared across lake, warehouse, and ingestion."""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Any
+
+VALID_OPS = frozenset({"I", "U", "D"})
+
+
+@dataclass(frozen=True)
+class CDCEvent:
+    event_id: int
+    source_table: str
+    op: str
+    pk_json: str
+    row_json: str
+    event_ts: datetime
+
+    def __post_init__(self) -> None:
+        if self.op not in VALID_OPS:
+            raise ValueError(
+                f"Invalid op {self.op!r}; expected one of {sorted(VALID_OPS)}"
+            )
+
+    @property
+    def primary_key(self) -> dict[str, Any]:
+        return json.loads(self.pk_json)
+
+    @property
+    def row(self) -> dict[str, Any]:
+        return json.loads(self.row_json)
+
+    @classmethod
+    def from_mysql_row(cls, row: dict[str, Any]) -> CDCEvent:
+        return cls(
+            event_id=int(row["event_id"]),
+            source_table=str(row["source_table"]),
+            op=str(row["op"]),
+            pk_json=_to_json_str(row["pk_json"]),
+            row_json=_to_json_str(row["row_json"]),
+            event_ts=row["event_ts"],
+        )
+
+
+def _to_json_str(value: Any) -> str:
+    if isinstance(value, str):
+        return value
+    if isinstance(value, (bytes, bytearray)):
+        return value.decode("utf-8")
+    return json.dumps(value, default=str)