From b23268d9368a30d7c78a67b9458acdb23e9010f8 Mon Sep 17 00:00:00 2001 From: Shravani More Date: Wed, 27 May 2026 18:04:26 +0530 Subject: [PATCH 1/4] docs: add initial setup and instructions --- submission/cdc-lakehouse/.gitignore | 8 ++ submission/cdc-lakehouse/README.md | 88 +++++++++++++++++++++ submission/cdc-lakehouse/docker-compose.yml | 21 +++++ submission/cdc-lakehouse/requirements.txt | 4 + 4 files changed, 121 insertions(+) create mode 100644 submission/cdc-lakehouse/.gitignore create mode 100644 submission/cdc-lakehouse/README.md create mode 100644 submission/cdc-lakehouse/docker-compose.yml create mode 100644 submission/cdc-lakehouse/requirements.txt diff --git a/submission/cdc-lakehouse/.gitignore b/submission/cdc-lakehouse/.gitignore new file mode 100644 index 0000000..37e92f1 --- /dev/null +++ b/submission/cdc-lakehouse/.gitignore @@ -0,0 +1,8 @@ +lake/ +warehouse/ +state/ +ingest/FAILED_SCHEMA_CHANGE.md +__pycache__/ +*.pyc +.pytest_cache/ +.venv/ diff --git a/submission/cdc-lakehouse/README.md b/submission/cdc-lakehouse/README.md new file mode 100644 index 0000000..2962826 --- /dev/null +++ b/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. diff --git a/submission/cdc-lakehouse/docker-compose.yml b/submission/cdc-lakehouse/docker-compose.yml new file mode 100644 index 0000000..dd3397b --- /dev/null +++ b/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 diff --git a/submission/cdc-lakehouse/requirements.txt b/submission/cdc-lakehouse/requirements.txt new file mode 100644 index 0000000..9ec2239 --- /dev/null +++ b/submission/cdc-lakehouse/requirements.txt @@ -0,0 +1,4 @@ +duckdb>=1.0.0 +pymysql>=1.1.0 +pyarrow>=15.0.0 +pytest>=8.0.0 From b598f65c9a2f08c0534f1645e6e8fc1d9957e36f Mon Sep 17 00:00:00 2001 From: Shravani More Date: Wed, 27 May 2026 18:04:36 +0530 Subject: [PATCH 2/4] feat: implement source schema, triggers, and schema contracts --- .../contracts/schema_contract.json | 58 +++++ .../contracts/schema_contract_spec.json | 75 ++++++ .../contracts/schema_fingerprint.json | 8 + .../cdc-lakehouse/source/install-triggers.sh | 3 + submission/cdc-lakehouse/source/models.py | 142 +++++++++++ submission/cdc-lakehouse/source/schema.sql | 107 ++++++++ submission/cdc-lakehouse/source/seed.sql | 27 ++ submission/cdc-lakehouse/source/triggers.sql | 230 ++++++++++++++++++ 8 files changed, 650 insertions(+) create mode 100644 submission/cdc-lakehouse/contracts/schema_contract.json create mode 100644 submission/cdc-lakehouse/contracts/schema_contract_spec.json create mode 100644 submission/cdc-lakehouse/contracts/schema_fingerprint.json create mode 100755 submission/cdc-lakehouse/source/install-triggers.sh create mode 100644 submission/cdc-lakehouse/source/models.py create mode 100644 submission/cdc-lakehouse/source/schema.sql create mode 100644 submission/cdc-lakehouse/source/seed.sql create mode 100644 submission/cdc-lakehouse/source/triggers.sql diff --git a/submission/cdc-lakehouse/contracts/schema_contract.json b/submission/cdc-lakehouse/contracts/schema_contract.json new file mode 100644 index 0000000..159835a --- /dev/null +++ b/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" + ] +} \ No newline at end of file diff --git a/submission/cdc-lakehouse/contracts/schema_contract_spec.json b/submission/cdc-lakehouse/contracts/schema_contract_spec.json new file mode 100644 index 0000000..33a4ca1 --- /dev/null +++ b/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": ""} + ] + } + } +} diff --git a/submission/cdc-lakehouse/contracts/schema_fingerprint.json b/submission/cdc-lakehouse/contracts/schema_fingerprint.json new file mode 100644 index 0000000..faacb35 --- /dev/null +++ b/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" +} \ No newline at end of file diff --git a/submission/cdc-lakehouse/source/install-triggers.sh b/submission/cdc-lakehouse/source/install-triggers.sh new file mode 100755 index 0000000..a24b2ba --- /dev/null +++ b/submission/cdc-lakehouse/source/install-triggers.sh @@ -0,0 +1,3 @@ +#!/bin/bash +set -e +mysql -uroot -p"${MYSQL_ROOT_PASSWORD}" "${MYSQL_DATABASE}" < /docker-entrypoint-initdb.d/98-triggers.sql diff --git a/submission/cdc-lakehouse/source/models.py b/submission/cdc-lakehouse/source/models.py new file mode 100644 index 0000000..2818d1f --- /dev/null +++ b/submission/cdc-lakehouse/source/models.py @@ -0,0 +1,142 @@ +""" +DuckDB mirror of the MySQL source schema for local tests and CI contract checks. + +Production source of record: MySQL (see source/schema.sql). +""" + +from __future__ import annotations + +import duckdb + +SCHEMA_CONTRACT: dict[str, list[str]] = { + "customers": [ + "customer_id", + "name", + "email", + "phone", + "status", + "created_at", + "updated_at", + "is_deleted", + ], + "products": [ + "product_id", + "sku", + "name", + "unit_price", + "created_at", + "updated_at", + "is_deleted", + ], + "orders": [ + "order_id", + "customer_id", + "order_status", + "order_total", + "created_at", + "updated_at", + "paid_at", + "cancel_reason", + "is_deleted", + ], + "order_items": [ + "order_item_id", + "order_id", + "product_id", + "quantity", + "unit_price", + "line_total", + "is_deleted", + ], + "payments": [ + "payment_id", + "order_id", + "payment_amount", + "payment_status", + "created_at", + "updated_at", + "is_deleted", + ], + "order_status_events": [ + "status_event_id", + "order_id", + "from_status", + "to_status", + "event_at", + "is_deleted", + ], +} + +BUSINESS_TABLES = list(SCHEMA_CONTRACT.keys()) + + +def create_source_tables(conn: duckdb.DuckDBPyConnection) -> None: + """Create in-memory source tables matching MySQL column contracts.""" + conn.execute(""" + CREATE TABLE customers ( + customer_id BIGINT PRIMARY KEY, + name VARCHAR NOT NULL, + email VARCHAR, + phone VARCHAR, + status VARCHAR NOT NULL, + created_at TIMESTAMP NOT NULL, + updated_at TIMESTAMP NOT NULL, + is_deleted BOOLEAN NOT NULL DEFAULT false + ) + """) + conn.execute(""" + CREATE TABLE products ( + product_id BIGINT PRIMARY KEY, + sku VARCHAR NOT NULL, + name VARCHAR NOT NULL, + unit_price DECIMAL(12, 2) NOT NULL, + created_at TIMESTAMP NOT NULL, + updated_at TIMESTAMP NOT NULL, + is_deleted BOOLEAN NOT NULL DEFAULT false + ) + """) + conn.execute(""" + CREATE TABLE orders ( + order_id BIGINT PRIMARY KEY, + customer_id BIGINT NOT NULL, + order_status VARCHAR NOT NULL, + order_total DECIMAL(12, 2) NOT NULL, + created_at TIMESTAMP NOT NULL, + updated_at TIMESTAMP NOT NULL, + paid_at TIMESTAMP, + cancel_reason VARCHAR, + is_deleted BOOLEAN NOT NULL DEFAULT false + ) + """) + conn.execute(""" + CREATE TABLE order_items ( + order_item_id BIGINT PRIMARY KEY, + order_id BIGINT NOT NULL, + product_id BIGINT NOT NULL, + quantity INTEGER NOT NULL, + unit_price DECIMAL(12, 2) NOT NULL, + line_total DECIMAL(12, 2) NOT NULL, + is_deleted BOOLEAN NOT NULL DEFAULT false + ) + """) + conn.execute(""" + CREATE TABLE payments ( + payment_id BIGINT PRIMARY KEY, + order_id BIGINT NOT NULL, + payment_amount DECIMAL(12, 2) NOT NULL, + payment_status VARCHAR NOT NULL, + created_at TIMESTAMP NOT NULL, + updated_at TIMESTAMP NOT NULL, + is_deleted BOOLEAN NOT NULL DEFAULT false + ) + """) + conn.execute(""" + CREATE TABLE order_status_events ( + status_event_id BIGINT PRIMARY KEY, + order_id BIGINT NOT NULL, + from_status VARCHAR, + to_status VARCHAR NOT NULL, + event_at TIMESTAMP NOT NULL, + is_deleted BOOLEAN NOT NULL DEFAULT false + ) + """) diff --git a/submission/cdc-lakehouse/source/schema.sql b/submission/cdc-lakehouse/source/schema.sql new file mode 100644 index 0000000..495213a --- /dev/null +++ b/submission/cdc-lakehouse/source/schema.sql @@ -0,0 +1,107 @@ +-- E-commerce source schema (MySQL 8) +-- Strong: customers, products, orders, payments +-- Weak: order_items, order_status_events + +CREATE DATABASE IF NOT EXISTS cdc_source; +USE cdc_source; + +-- --------------------------------------------------------------------------- +-- CDC outbox (simulates binlog / Debezium in production) +-- --------------------------------------------------------------------------- +CREATE TABLE IF NOT EXISTS cdc_events ( + event_id BIGINT AUTO_INCREMENT PRIMARY KEY, + source_table VARCHAR(64) NOT NULL, + op CHAR(1) NOT NULL, + pk_json JSON NOT NULL, + row_json JSON NOT NULL, + event_ts TIMESTAMP(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6), + INDEX idx_cdc_events_event_id (event_id), + INDEX idx_cdc_events_source_table (source_table) +); + +-- --------------------------------------------------------------------------- +-- Business tables +-- --------------------------------------------------------------------------- +CREATE TABLE IF NOT EXISTS customers ( + customer_id BIGINT AUTO_INCREMENT PRIMARY KEY, + name VARCHAR(255) NOT NULL, + email VARCHAR(255) NULL, + phone VARCHAR(50) NULL, + status ENUM('ACTIVE', 'SUSPENDED', 'CLOSED') NOT NULL DEFAULT 'ACTIVE', + created_at TIMESTAMP(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6), + updated_at TIMESTAMP(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6), + is_deleted TINYINT(1) NOT NULL DEFAULT 0, + INDEX idx_customers_updated_at (updated_at), + INDEX idx_customers_status (status) +); + +CREATE TABLE IF NOT EXISTS products ( + product_id BIGINT AUTO_INCREMENT PRIMARY KEY, + sku VARCHAR(64) NOT NULL, + name VARCHAR(255) NOT NULL, + unit_price DECIMAL(12, 2) NOT NULL, + created_at TIMESTAMP(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6), + updated_at TIMESTAMP(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6), + is_deleted TINYINT(1) NOT NULL DEFAULT 0, + UNIQUE KEY uk_products_sku (sku), + INDEX idx_products_updated_at (updated_at), + CONSTRAINT chk_products_unit_price CHECK (unit_price >= 0) +); + +CREATE TABLE IF NOT EXISTS orders ( + order_id BIGINT AUTO_INCREMENT PRIMARY KEY, + customer_id BIGINT NOT NULL, + order_status ENUM('CREATED', 'PAID', 'SHIPPED', 'CANCELLED', 'REFUNDED') NOT NULL DEFAULT 'CREATED', + order_total DECIMAL(12, 2) NOT NULL DEFAULT 0, + created_at TIMESTAMP(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6), + updated_at TIMESTAMP(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6), + paid_at TIMESTAMP(6) NULL, + cancel_reason VARCHAR(512) NULL, + is_deleted TINYINT(1) NOT NULL DEFAULT 0, + INDEX idx_orders_customer_id (customer_id), + INDEX idx_orders_updated_at (updated_at), + INDEX idx_orders_status (order_status), + CONSTRAINT fk_orders_customer FOREIGN KEY (customer_id) REFERENCES customers (customer_id), + CONSTRAINT chk_orders_total CHECK (order_total >= 0) +); + +CREATE TABLE IF NOT EXISTS order_items ( + order_item_id BIGINT AUTO_INCREMENT PRIMARY KEY, + order_id BIGINT NOT NULL, + product_id BIGINT NOT NULL, + quantity INT NOT NULL DEFAULT 1, + unit_price DECIMAL(12, 2) NOT NULL, + line_total DECIMAL(12, 2) NOT NULL, + is_deleted TINYINT(1) NOT NULL DEFAULT 0, + INDEX idx_order_items_order_id (order_id), + INDEX idx_order_items_product_id (product_id), + CONSTRAINT fk_order_items_order FOREIGN KEY (order_id) REFERENCES orders (order_id), + CONSTRAINT fk_order_items_product FOREIGN KEY (product_id) REFERENCES products (product_id), + CONSTRAINT chk_order_items_qty CHECK (quantity > 0), + CONSTRAINT chk_order_items_prices CHECK (unit_price >= 0 AND line_total >= 0) +); + +CREATE TABLE IF NOT EXISTS payments ( + payment_id BIGINT AUTO_INCREMENT PRIMARY KEY, + order_id BIGINT NOT NULL, + payment_amount DECIMAL(12, 2) NOT NULL, + payment_status ENUM('PENDING', 'CAPTURED', 'FAILED', 'REFUNDED') NOT NULL DEFAULT 'PENDING', + created_at TIMESTAMP(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6), + updated_at TIMESTAMP(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6), + is_deleted TINYINT(1) NOT NULL DEFAULT 0, + INDEX idx_payments_order_id (order_id), + INDEX idx_payments_updated_at (updated_at), + CONSTRAINT fk_payments_order FOREIGN KEY (order_id) REFERENCES orders (order_id), + CONSTRAINT chk_payments_amount CHECK (payment_amount >= 0) +); + +CREATE TABLE IF NOT EXISTS order_status_events ( + status_event_id BIGINT AUTO_INCREMENT PRIMARY KEY, + order_id BIGINT NOT NULL, + from_status ENUM('CREATED', 'PAID', 'SHIPPED', 'CANCELLED', 'REFUNDED') NULL, + to_status ENUM('CREATED', 'PAID', 'SHIPPED', 'CANCELLED', 'REFUNDED') NOT NULL, + event_at TIMESTAMP(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6), + is_deleted TINYINT(1) NOT NULL DEFAULT 0, + INDEX idx_status_events_order_id (order_id), + CONSTRAINT fk_status_events_order FOREIGN KEY (order_id) REFERENCES orders (order_id) +); diff --git a/submission/cdc-lakehouse/source/seed.sql b/submission/cdc-lakehouse/source/seed.sql new file mode 100644 index 0000000..1c2fc0c --- /dev/null +++ b/submission/cdc-lakehouse/source/seed.sql @@ -0,0 +1,27 @@ +USE cdc_source; + +INSERT INTO customers (name, email, status) VALUES + ('Alice Example', 'alice@example.com', 'ACTIVE'), + ('Bob Example', 'bob@example.com', 'ACTIVE'); + +INSERT INTO products (sku, name, unit_price) VALUES + ('SKU-001', 'Widget A', 19.99), + ('SKU-002', 'Widget B', 29.50); + +INSERT INTO orders (customer_id, order_status, order_total) VALUES + (1, 'CREATED', 39.98); + +INSERT INTO order_items (order_id, product_id, quantity, unit_price, line_total) VALUES + (1, 1, 2, 19.99, 39.98); + +INSERT INTO order_status_events (order_id, from_status, to_status) VALUES + (1, NULL, 'CREATED'); + +INSERT INTO payments (order_id, payment_amount, payment_status) VALUES + (1, 39.98, 'PENDING'); + +UPDATE orders SET order_status = 'PAID', paid_at = CURRENT_TIMESTAMP(6) WHERE order_id = 1; +UPDATE payments SET payment_status = 'CAPTURED' WHERE payment_id = 1; + +INSERT INTO order_status_events (order_id, from_status, to_status) VALUES + (1, 'CREATED', 'PAID'); diff --git a/submission/cdc-lakehouse/source/triggers.sql b/submission/cdc-lakehouse/source/triggers.sql new file mode 100644 index 0000000..6207759 --- /dev/null +++ b/submission/cdc-lakehouse/source/triggers.sql @@ -0,0 +1,230 @@ +USE cdc_source; + +DELIMITER $$ + +-- customers +CREATE TRIGGER trg_customers_ai AFTER INSERT ON customers FOR EACH ROW +BEGIN + INSERT INTO cdc_events (source_table, op, pk_json, row_json) + VALUES ( + 'customers', 'I', + JSON_OBJECT('customer_id', NEW.customer_id), + JSON_OBJECT('customer_id', NEW.customer_id, 'name', NEW.name, 'email', NEW.email, + 'phone', NEW.phone, 'status', NEW.status, 'created_at', NEW.created_at, + 'updated_at', NEW.updated_at, 'is_deleted', NEW.is_deleted) + ); +END$$ + +CREATE TRIGGER trg_customers_au AFTER UPDATE ON customers FOR EACH ROW +BEGIN + INSERT INTO cdc_events (source_table, op, pk_json, row_json) + VALUES ( + 'customers', 'U', + JSON_OBJECT('customer_id', NEW.customer_id), + JSON_OBJECT('customer_id', NEW.customer_id, 'name', NEW.name, 'email', NEW.email, + 'phone', NEW.phone, 'status', NEW.status, 'created_at', NEW.created_at, + 'updated_at', NEW.updated_at, 'is_deleted', NEW.is_deleted) + ); +END$$ + +CREATE TRIGGER trg_customers_ad AFTER DELETE ON customers FOR EACH ROW +BEGIN + INSERT INTO cdc_events (source_table, op, pk_json, row_json) + VALUES ( + 'customers', 'D', + JSON_OBJECT('customer_id', OLD.customer_id), + JSON_OBJECT('customer_id', OLD.customer_id, 'name', OLD.name, 'email', OLD.email, + 'phone', OLD.phone, 'status', OLD.status, 'created_at', OLD.created_at, + 'updated_at', OLD.updated_at, 'is_deleted', OLD.is_deleted) + ); +END$$ + +-- products +CREATE TRIGGER trg_products_ai AFTER INSERT ON products FOR EACH ROW +BEGIN + INSERT INTO cdc_events (source_table, op, pk_json, row_json) + VALUES ( + 'products', 'I', + JSON_OBJECT('product_id', NEW.product_id), + JSON_OBJECT('product_id', NEW.product_id, 'sku', NEW.sku, 'name', NEW.name, + 'unit_price', NEW.unit_price, 'created_at', NEW.created_at, + 'updated_at', NEW.updated_at, 'is_deleted', NEW.is_deleted) + ); +END$$ + +CREATE TRIGGER trg_products_au AFTER UPDATE ON products FOR EACH ROW +BEGIN + INSERT INTO cdc_events (source_table, op, pk_json, row_json) + VALUES ( + 'products', 'U', + JSON_OBJECT('product_id', NEW.product_id), + JSON_OBJECT('product_id', NEW.product_id, 'sku', NEW.sku, 'name', NEW.name, + 'unit_price', NEW.unit_price, 'created_at', NEW.created_at, + 'updated_at', NEW.updated_at, 'is_deleted', NEW.is_deleted) + ); +END$$ + +CREATE TRIGGER trg_products_ad AFTER DELETE ON products FOR EACH ROW +BEGIN + INSERT INTO cdc_events (source_table, op, pk_json, row_json) + VALUES ( + 'products', 'D', + JSON_OBJECT('product_id', OLD.product_id), + JSON_OBJECT('product_id', OLD.product_id, 'sku', OLD.sku, 'name', OLD.name, + 'unit_price', OLD.unit_price, 'created_at', OLD.created_at, + 'updated_at', OLD.updated_at, 'is_deleted', OLD.is_deleted) + ); +END$$ + +-- orders +CREATE TRIGGER trg_orders_ai AFTER INSERT ON orders FOR EACH ROW +BEGIN + INSERT INTO cdc_events (source_table, op, pk_json, row_json) + VALUES ( + 'orders', 'I', + JSON_OBJECT('order_id', NEW.order_id), + JSON_OBJECT('order_id', NEW.order_id, 'customer_id', NEW.customer_id, + 'order_status', NEW.order_status, 'order_total', NEW.order_total, + 'created_at', NEW.created_at, 'updated_at', NEW.updated_at, + 'paid_at', NEW.paid_at, 'cancel_reason', NEW.cancel_reason, 'is_deleted', NEW.is_deleted) + ); +END$$ + +CREATE TRIGGER trg_orders_au AFTER UPDATE ON orders FOR EACH ROW +BEGIN + INSERT INTO cdc_events (source_table, op, pk_json, row_json) + VALUES ( + 'orders', 'U', + JSON_OBJECT('order_id', NEW.order_id), + JSON_OBJECT('order_id', NEW.order_id, 'customer_id', NEW.customer_id, + 'order_status', NEW.order_status, 'order_total', NEW.order_total, + 'created_at', NEW.created_at, 'updated_at', NEW.updated_at, + 'paid_at', NEW.paid_at, 'cancel_reason', NEW.cancel_reason, 'is_deleted', NEW.is_deleted) + ); +END$$ + +CREATE TRIGGER trg_orders_ad AFTER DELETE ON orders FOR EACH ROW +BEGIN + INSERT INTO cdc_events (source_table, op, pk_json, row_json) + VALUES ( + 'orders', 'D', + JSON_OBJECT('order_id', OLD.order_id), + JSON_OBJECT('order_id', OLD.order_id, 'customer_id', OLD.customer_id, + 'order_status', OLD.order_status, 'order_total', OLD.order_total, + 'created_at', OLD.created_at, 'updated_at', OLD.updated_at, + 'paid_at', OLD.paid_at, 'cancel_reason', OLD.cancel_reason, 'is_deleted', OLD.is_deleted) + ); +END$$ + +-- order_items +CREATE TRIGGER trg_order_items_ai AFTER INSERT ON order_items FOR EACH ROW +BEGIN + INSERT INTO cdc_events (source_table, op, pk_json, row_json) + VALUES ( + 'order_items', 'I', + JSON_OBJECT('order_item_id', NEW.order_item_id), + JSON_OBJECT('order_item_id', NEW.order_item_id, 'order_id', NEW.order_id, + 'product_id', NEW.product_id, 'quantity', NEW.quantity, 'unit_price', NEW.unit_price, + 'line_total', NEW.line_total, 'is_deleted', NEW.is_deleted) + ); +END$$ + +CREATE TRIGGER trg_order_items_au AFTER UPDATE ON order_items FOR EACH ROW +BEGIN + INSERT INTO cdc_events (source_table, op, pk_json, row_json) + VALUES ( + 'order_items', 'U', + JSON_OBJECT('order_item_id', NEW.order_item_id), + JSON_OBJECT('order_item_id', NEW.order_item_id, 'order_id', NEW.order_id, + 'product_id', NEW.product_id, 'quantity', NEW.quantity, 'unit_price', NEW.unit_price, + 'line_total', NEW.line_total, 'is_deleted', NEW.is_deleted) + ); +END$$ + +CREATE TRIGGER trg_order_items_ad AFTER DELETE ON order_items FOR EACH ROW +BEGIN + INSERT INTO cdc_events (source_table, op, pk_json, row_json) + VALUES ( + 'order_items', 'D', + JSON_OBJECT('order_item_id', OLD.order_item_id), + JSON_OBJECT('order_item_id', OLD.order_item_id, 'order_id', OLD.order_id, + 'product_id', OLD.product_id, 'quantity', OLD.quantity, 'unit_price', OLD.unit_price, + 'line_total', OLD.line_total, 'is_deleted', OLD.is_deleted) + ); +END$$ + +-- payments +CREATE TRIGGER trg_payments_ai AFTER INSERT ON payments FOR EACH ROW +BEGIN + INSERT INTO cdc_events (source_table, op, pk_json, row_json) + VALUES ( + 'payments', 'I', + JSON_OBJECT('payment_id', NEW.payment_id), + JSON_OBJECT('payment_id', NEW.payment_id, 'order_id', NEW.order_id, + 'payment_amount', NEW.payment_amount, 'payment_status', NEW.payment_status, + 'created_at', NEW.created_at, 'updated_at', NEW.updated_at, 'is_deleted', NEW.is_deleted) + ); +END$$ + +CREATE TRIGGER trg_payments_au AFTER UPDATE ON payments FOR EACH ROW +BEGIN + INSERT INTO cdc_events (source_table, op, pk_json, row_json) + VALUES ( + 'payments', 'U', + JSON_OBJECT('payment_id', NEW.payment_id), + JSON_OBJECT('payment_id', NEW.payment_id, 'order_id', NEW.order_id, + 'payment_amount', NEW.payment_amount, 'payment_status', NEW.payment_status, + 'created_at', NEW.created_at, 'updated_at', NEW.updated_at, 'is_deleted', NEW.is_deleted) + ); +END$$ + +CREATE TRIGGER trg_payments_ad AFTER DELETE ON payments FOR EACH ROW +BEGIN + INSERT INTO cdc_events (source_table, op, pk_json, row_json) + VALUES ( + 'payments', 'D', + JSON_OBJECT('payment_id', OLD.payment_id), + JSON_OBJECT('payment_id', OLD.payment_id, 'order_id', OLD.order_id, + 'payment_amount', OLD.payment_amount, 'payment_status', OLD.payment_status, + 'created_at', OLD.created_at, 'updated_at', OLD.updated_at, 'is_deleted', OLD.is_deleted) + ); +END$$ + +-- order_status_events +CREATE TRIGGER trg_status_events_ai AFTER INSERT ON order_status_events FOR EACH ROW +BEGIN + INSERT INTO cdc_events (source_table, op, pk_json, row_json) + VALUES ( + 'order_status_events', 'I', + JSON_OBJECT('status_event_id', NEW.status_event_id), + JSON_OBJECT('status_event_id', NEW.status_event_id, 'order_id', NEW.order_id, + 'from_status', NEW.from_status, 'to_status', NEW.to_status, + 'event_at', NEW.event_at, 'is_deleted', NEW.is_deleted) + ); +END$$ + +CREATE TRIGGER trg_status_events_au AFTER UPDATE ON order_status_events FOR EACH ROW +BEGIN + INSERT INTO cdc_events (source_table, op, pk_json, row_json) + VALUES ( + 'order_status_events', 'U', + JSON_OBJECT('status_event_id', NEW.status_event_id), + JSON_OBJECT('status_event_id', NEW.status_event_id, 'order_id', NEW.order_id, + 'from_status', NEW.from_status, 'to_status', NEW.to_status, + 'event_at', NEW.event_at, 'is_deleted', NEW.is_deleted) + ); +END$$ + +CREATE TRIGGER trg_status_events_ad AFTER DELETE ON order_status_events FOR EACH ROW +BEGIN + INSERT INTO cdc_events (source_table, op, pk_json, row_json) + VALUES ( + 'order_status_events', 'D', + JSON_OBJECT('status_event_id', OLD.status_event_id), + JSON_OBJECT('status_event_id', OLD.status_event_id, 'order_id', OLD.order_id, + 'from_status', OLD.from_status, 'to_status', OLD.to_status, + 'event_at', OLD.event_at, 'is_deleted', OLD.is_deleted) + ); +END$$ + +DELIMITER ; From e6a0b57e49d6dde579691da11be73473524049b8 Mon Sep 17 00:00:00 2001 From: Shravani More Date: Wed, 27 May 2026 18:05:01 +0530 Subject: [PATCH 3/4] feat: implement cdc ingestion pipeline and scripts --- submission/cdc-lakehouse/pipeline/__init__.py | 1 + .../cdc-lakehouse/pipeline/cdc_event.py | 53 ++++ .../cdc-lakehouse/pipeline/checkpoint.py | 24 ++ submission/cdc-lakehouse/pipeline/config.py | 27 ++ submission/cdc-lakehouse/pipeline/ingest.py | 76 +++++ .../cdc-lakehouse/pipeline/lake_writer.py | 104 +++++++ .../cdc-lakehouse/pipeline/mysql_client.py | 66 +++++ .../cdc-lakehouse/pipeline/schema_check.py | 260 ++++++++++++++++++ .../cdc-lakehouse/pipeline/warehouse_apply.py | 170 ++++++++++++ .../scripts/check_schema_contracts.py | 43 +++ .../scripts/compile_schema_fingerprint.py | 34 +++ .../scripts/restore_warehouse.py | 67 +++++ .../scripts/run_data_quality_checks.py | 123 +++++++++ .../cdc-lakehouse/scripts/run_ingestion.sh | 5 + .../cdc-lakehouse/scripts/validate_catalog.py | 60 ++++ 15 files changed, 1113 insertions(+) create mode 100644 submission/cdc-lakehouse/pipeline/__init__.py create mode 100644 submission/cdc-lakehouse/pipeline/cdc_event.py create mode 100644 submission/cdc-lakehouse/pipeline/checkpoint.py create mode 100644 submission/cdc-lakehouse/pipeline/config.py create mode 100644 submission/cdc-lakehouse/pipeline/ingest.py create mode 100644 submission/cdc-lakehouse/pipeline/lake_writer.py create mode 100644 submission/cdc-lakehouse/pipeline/mysql_client.py create mode 100644 submission/cdc-lakehouse/pipeline/schema_check.py create mode 100644 submission/cdc-lakehouse/pipeline/warehouse_apply.py create mode 100644 submission/cdc-lakehouse/scripts/check_schema_contracts.py create mode 100644 submission/cdc-lakehouse/scripts/compile_schema_fingerprint.py create mode 100644 submission/cdc-lakehouse/scripts/restore_warehouse.py create mode 100644 submission/cdc-lakehouse/scripts/run_data_quality_checks.py create mode 100755 submission/cdc-lakehouse/scripts/run_ingestion.sh create mode 100644 submission/cdc-lakehouse/scripts/validate_catalog.py diff --git a/submission/cdc-lakehouse/pipeline/__init__.py b/submission/cdc-lakehouse/pipeline/__init__.py new file mode 100644 index 0000000..168eebc --- /dev/null +++ b/submission/cdc-lakehouse/pipeline/__init__.py @@ -0,0 +1 @@ +"""CDC lakehouse pipeline package.""" diff --git a/submission/cdc-lakehouse/pipeline/cdc_event.py b/submission/cdc-lakehouse/pipeline/cdc_event.py new file mode 100644 index 0000000..eb7a05c --- /dev/null +++ b/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) diff --git a/submission/cdc-lakehouse/pipeline/checkpoint.py b/submission/cdc-lakehouse/pipeline/checkpoint.py new file mode 100644 index 0000000..4fbf616 --- /dev/null +++ b/submission/cdc-lakehouse/pipeline/checkpoint.py @@ -0,0 +1,24 @@ +"""Checkpoint persistence for CDC ingestion.""" + +from __future__ import annotations + +import json +from pathlib import Path + +from pipeline.config import CHECKPOINT_PATH, STATE_DIR + + +def load_checkpoint(path: Path | None = None) -> int: + cp_path = path or CHECKPOINT_PATH + if not cp_path.exists(): + return 0 + data = json.loads(cp_path.read_text(encoding="utf-8")) + return int(data.get("last_event_id", 0)) + + +def save_checkpoint(last_event_id: int, path: Path | None = None) -> None: + cp_path = path or CHECKPOINT_PATH + STATE_DIR.mkdir(parents=True, exist_ok=True) + cp_path.write_text( + json.dumps({"last_event_id": last_event_id}, indent=2), encoding="utf-8" + ) diff --git a/submission/cdc-lakehouse/pipeline/config.py b/submission/cdc-lakehouse/pipeline/config.py new file mode 100644 index 0000000..88d72d3 --- /dev/null +++ b/submission/cdc-lakehouse/pipeline/config.py @@ -0,0 +1,27 @@ +"""Paths and environment configuration.""" + +from __future__ import annotations + +import os +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +LAKE_DIR = Path(os.environ.get("LAKE_DIR", ROOT / "lake")) +WAREHOUSE_PATH = Path( + os.environ.get("WAREHOUSE_PATH", ROOT / "warehouse" / "warehouse.duckdb") +) +STATE_DIR = Path(os.environ.get("STATE_DIR", ROOT / "state")) +CHECKPOINT_PATH = STATE_DIR / "checkpoint.json" +CONTRACTS_DIR = ROOT / "contracts" +SCHEMA_CONTRACT_PATH = CONTRACTS_DIR / "schema_contract.json" +SCHEMA_CONTRACT_SPEC_PATH = CONTRACTS_DIR / "schema_contract_spec.json" +SCHEMA_FINGERPRINT_PATH = CONTRACTS_DIR / "schema_fingerprint.json" +FAILED_SCHEMA_PATH = ROOT / "ingest" / "FAILED_SCHEMA_CHANGE.md" + +MYSQL_CONFIG = { + "host": os.environ.get("MYSQL_HOST", "127.0.0.1"), + "port": int(os.environ.get("MYSQL_PORT", "3306")), + "user": os.environ.get("MYSQL_USER", "cdc_user"), + "password": os.environ.get("MYSQL_PASSWORD", "cdc_pass"), + "database": os.environ.get("MYSQL_DATABASE", "cdc_source"), +} diff --git a/submission/cdc-lakehouse/pipeline/ingest.py b/submission/cdc-lakehouse/pipeline/ingest.py new file mode 100644 index 0000000..172aad9 --- /dev/null +++ b/submission/cdc-lakehouse/pipeline/ingest.py @@ -0,0 +1,76 @@ +#!/usr/bin/env python3 +"""Run CDC ingestion: schema check → lake → warehouse → checkpoint.""" + +from __future__ import annotations + +import sys + +import duckdb + +from pipeline.checkpoint import load_checkpoint, save_checkpoint +from pipeline.config import WAREHOUSE_PATH +from pipeline.lake_writer import write_events_to_lake +from pipeline.mysql_client import ( + connect_mysql, + fetch_cdc_events_since, + fetch_information_schema, +) +from pipeline.schema_check import ( + export_contract_file, + validate_schema, + write_failure_artifact, +) +from pipeline.warehouse_apply import apply_events, create_warehouse_tables + + +def run_ingestion() -> int: + export_contract_file() + + try: + mysql = connect_mysql() + except Exception as exc: + print(f"Cannot connect to MySQL: {exc}", file=sys.stderr) + return 1 + + info_cols = fetch_information_schema(mysql) + + # Always validate live source against source-team contract before any ingest. + violations = validate_schema(info_schema_columns=info_cols) + if violations: + write_failure_artifact(violations) + print( + "Schema validation failed — ingestion stopped (live source ≠ source contract).", + file=sys.stderr, + ) + for v in violations: + print(f" ✗ {v}", file=sys.stderr) + return 1 + + print("Schema validation passed (live source matches source contract).") + + last_event_id = load_checkpoint() + events = fetch_cdc_events_since(mysql, last_event_id) + if not events: + print(f"No new CDC events after event_id={last_event_id}.") + return 0 + + write_events_to_lake(events) + + WAREHOUSE_PATH.parent.mkdir(parents=True, exist_ok=True) + wh_conn = duckdb.connect(str(WAREHOUSE_PATH)) + create_warehouse_tables(wh_conn) + apply_events(wh_conn, events) + wh_conn.close() + + new_checkpoint = max(e.event_id for e in events) + save_checkpoint(new_checkpoint) + print(f"Ingested {len(events)} events; checkpoint={new_checkpoint}.") + return 0 + + +def main() -> int: + return run_ingestion() + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/submission/cdc-lakehouse/pipeline/lake_writer.py b/submission/cdc-lakehouse/pipeline/lake_writer.py new file mode 100644 index 0000000..fe5d458 --- /dev/null +++ b/submission/cdc-lakehouse/pipeline/lake_writer.py @@ -0,0 +1,104 @@ +"""Append-only Parquet lake writer for CDC events.""" + +from __future__ import annotations + +import uuid +from datetime import datetime, timezone +from pathlib import Path + +import pyarrow as pa +import pyarrow.parquet as pq + +from pipeline import config +from pipeline.cdc_event import CDCEvent + +LAKE_SCHEMA = pa.schema( + [ + ("event_id", pa.int64()), + ("source_table", pa.string()), + ("op", pa.string()), + ("pk_json", pa.string()), + ("row_json", pa.string()), + ("event_ts", pa.timestamp("us", tz="UTC")), + ("ingested_at", pa.timestamp("us", tz="UTC")), + ] +) + + +def _partition_path(source_table: str, event_ts: datetime, run_id: str) -> Path: + dt = ( + event_ts.astimezone(timezone.utc) + if event_ts.tzinfo + else event_ts.replace(tzinfo=timezone.utc) + ) + return ( + config.LAKE_DIR + / f"source_table={source_table}" + / f"dt={dt.date().isoformat()}" + / f"part-{run_id}.parquet" + ) + + +def write_events_to_lake( + events: list[CDCEvent], run_id: str | None = None +) -> list[Path]: + if not events: + return [] + + run_id = run_id or uuid.uuid4().hex[:12] + ingested_at = datetime.now(timezone.utc) + by_partition: dict[tuple[str, str], list[CDCEvent]] = {} + + for event in events: + dt_key = ( + event.event_ts.astimezone(timezone.utc).date().isoformat() + if event.event_ts.tzinfo + else event.event_ts.date().isoformat() + ) + key = (event.source_table, dt_key) + by_partition.setdefault(key, []).append(event) + + written: list[Path] = [] + for (source_table, _dt_key), partition_events in by_partition.items(): + path = _partition_path(source_table, partition_events[0].event_ts, run_id) + path.parent.mkdir(parents=True, exist_ok=True) + + table = pa.Table.from_pydict( + { + "event_id": [e.event_id for e in partition_events], + "source_table": [e.source_table for e in partition_events], + "op": [e.op for e in partition_events], + "pk_json": [e.pk_json for e in partition_events], + "row_json": [e.row_json for e in partition_events], + "event_ts": [e.event_ts for e in partition_events], + "ingested_at": [ingested_at] * len(partition_events), + }, + schema=LAKE_SCHEMA, + ) + pq.write_table(table, path) + written.append(path) + + return written + + +def read_all_lake_events(lake_dir: Path | None = None) -> list[CDCEvent]: + base = lake_dir or config.LAKE_DIR + if not base.exists(): + return [] + + events: list[CDCEvent] = [] + for parquet_file in sorted(base.glob("source_table=*/dt=*/*.parquet")): + table = pq.ParquetFile(parquet_file).read() + for row in table.to_pylist(): + events.append( + CDCEvent( + event_id=int(row["event_id"]), + source_table=str(row["source_table"]), + op=str(row["op"]), + pk_json=str(row["pk_json"]), + row_json=str(row["row_json"]), + event_ts=row["event_ts"], + ) + ) + events.sort(key=lambda e: e.event_id) + return events diff --git a/submission/cdc-lakehouse/pipeline/mysql_client.py b/submission/cdc-lakehouse/pipeline/mysql_client.py new file mode 100644 index 0000000..fd3554c --- /dev/null +++ b/submission/cdc-lakehouse/pipeline/mysql_client.py @@ -0,0 +1,66 @@ +"""MySQL connectivity for CDC ingestion.""" + +from __future__ import annotations + +from typing import Any + +import pymysql +from pymysql.cursors import DictCursor + +from pipeline.config import MYSQL_CONFIG +from pipeline.cdc_event import CDCEvent + + +def connect_mysql() -> pymysql.connections.Connection: + return pymysql.connect( + host=MYSQL_CONFIG["host"], + port=MYSQL_CONFIG["port"], + user=MYSQL_CONFIG["user"], + password=MYSQL_CONFIG["password"], + database=MYSQL_CONFIG["database"], + cursorclass=DictCursor, + autocommit=True, + ) + + +def fetch_cdc_events_since( + conn: pymysql.connections.Connection, last_event_id: int +) -> list[CDCEvent]: + with conn.cursor() as cur: + cur.execute( + """ + SELECT event_id, source_table, op, pk_json, row_json, event_ts + FROM cdc_events + WHERE event_id > %s + ORDER BY event_id ASC + """, + (last_event_id,), + ) + rows = cur.fetchall() + return [CDCEvent.from_mysql_row(row) for row in rows] + + +def fetch_information_schema( + conn: pymysql.connections.Connection, +) -> list[dict[str, Any]]: + tables = ( + "customers", + "products", + "orders", + "order_items", + "payments", + "order_status_events", + ) + placeholders = ", ".join(["%s"] * len(tables)) + with conn.cursor() as cur: + cur.execute( + f""" + SELECT table_name, column_name, data_type, is_nullable, column_key + FROM information_schema.columns + WHERE table_schema = %s + AND table_name IN ({placeholders}) + ORDER BY table_name, ordinal_position + """, + [MYSQL_CONFIG["database"], *tables], + ) + return list(cur.fetchall()) diff --git a/submission/cdc-lakehouse/pipeline/schema_check.py b/submission/cdc-lakehouse/pipeline/schema_check.py new file mode 100644 index 0000000..89236c4 --- /dev/null +++ b/submission/cdc-lakehouse/pipeline/schema_check.py @@ -0,0 +1,260 @@ +"""Schema contract validation — source-owned contract vs live MySQL on every ingest.""" + +from __future__ import annotations + +import hashlib +import json +from pathlib import Path +from typing import Any + +import duckdb + +from pipeline.config import ( + CONTRACTS_DIR, + FAILED_SCHEMA_PATH, + SCHEMA_CONTRACT_PATH, + SCHEMA_CONTRACT_SPEC_PATH, + SCHEMA_FINGERPRINT_PATH, +) + +# Legacy column-list contract (CI / DuckDB tests) +from source.models import SCHEMA_CONTRACT + + +def load_source_contract_spec() -> dict[str, Any]: + """Load the source-team schema contract (authoritative).""" + path = SCHEMA_CONTRACT_SPEC_PATH + if not path.exists(): + raise FileNotFoundError(f"Source contract missing: {path}") + return json.loads(path.read_text(encoding="utf-8")) + + +def contract_spec_to_information_schema_rows(spec: dict[str, Any]) -> list[dict[str, Any]]: + """Convert contract spec to INFORMATION_SCHEMA-shaped rows for fingerprinting.""" + rows: list[dict[str, Any]] = [] + for table, table_def in spec["tables"].items(): + for col in table_def["columns"]: + rows.append( + { + "TABLE_NAME": table, + "COLUMN_NAME": col["name"], + "DATA_TYPE": col["data_type"], + "IS_NULLABLE": "YES" if col["nullable"] else "NO", + "COLUMN_KEY": col.get("column_key", ""), + } + ) + return rows + + +def _fingerprint_from_columns(columns: list[dict[str, Any]]) -> dict[str, str]: + """Structural hash from column name, type, and nullability (not incidental indexes).""" + by_table: dict[str, list[str]] = {} + for col in columns: + table = col["table_name"] if "table_name" in col else col["TABLE_NAME"] + name = col["column_name"] if "column_name" in col else col["COLUMN_NAME"] + dtype = (col["data_type"] if "data_type" in col else col["DATA_TYPE"]).lower() + nullable = col["is_nullable"] if "is_nullable" in col else col["IS_NULLABLE"] + by_table.setdefault(table, []).append(f"{name}:{dtype}:{nullable}") + return { + t: hashlib.sha256("|".join(parts).encode()).hexdigest() + for t, parts in sorted(by_table.items()) + } + + +def fingerprint_from_mysql(columns: list[dict[str, Any]]) -> dict[str, str]: + normalized = [ + { + "TABLE_NAME": c["TABLE_NAME"] if "TABLE_NAME" in c else c["table_name"], + "COLUMN_NAME": c["COLUMN_NAME"] if "COLUMN_NAME" in c else c["column_name"], + "DATA_TYPE": c["DATA_TYPE"] if "DATA_TYPE" in c else c["data_type"], + "IS_NULLABLE": c["IS_NULLABLE"] if "IS_NULLABLE" in c else c["is_nullable"], + "COLUMN_KEY": c.get("COLUMN_KEY") or c.get("column_key") or "", + } + for c in columns + ] + return _fingerprint_from_columns(normalized) + + +def fingerprint_from_duckdb(conn: duckdb.DuckDBPyConnection) -> dict[str, str]: + columns: list[dict[str, Any]] = [] + for table in SCHEMA_CONTRACT: + rows = conn.execute(f"DESCRIBE {table}").fetchall() + for row in rows: + columns.append( + { + "table_name": table, + "column_name": row[0], + "data_type": row[1], + "is_nullable": row[2], + "column_key": row[3] if len(row) > 3 else "", + } + ) + return _fingerprint_from_columns(columns) + + +def expected_fingerprint_from_contract() -> dict[str, str]: + """Expected fingerprint derived from source contract — not from live DB.""" + spec = load_source_contract_spec() + return fingerprint_from_mysql(contract_spec_to_information_schema_rows(spec)) + + +def load_expected_fingerprint() -> dict[str, str]: + """Load committed expected fingerprint (compiled from source contract).""" + if SCHEMA_FINGERPRINT_PATH.exists(): + return json.loads(SCHEMA_FINGERPRINT_PATH.read_text(encoding="utf-8")) + # Fallback: compute from contract spec at validation time + return expected_fingerprint_from_contract() + + +def check_contract_columns(actual_by_table: dict[str, set[str]]) -> list[str]: + violations: list[str] = [] + for table, expected_cols in SCHEMA_CONTRACT.items(): + actual = actual_by_table.get(table, set()) + if not actual: + violations.append(f"{table}: table missing from source") + continue + for col in expected_cols: + if col not in actual: + violations.append(f"{table}.{col}: column missing from source table") + return violations + + +def validate_live_against_source_contract( + info_schema_columns: list[dict[str, Any]], +) -> list[str]: + """ + Validate live source schema against the source-team contract on every ingestion. + + Checks: required tables/columns, types, nullability, keys, no unexpected columns, + and structural fingerprint match to expected contract fingerprint. + """ + spec = load_source_contract_spec() + violations: list[str] = [] + + live_by_table: dict[str, dict[str, dict[str, Any]]] = {} + for col in info_schema_columns: + table = col.get("TABLE_NAME") or col.get("table_name") + name = col.get("COLUMN_NAME") or col.get("column_name") + live_by_table.setdefault(table, {})[name] = { + "data_type": (col.get("DATA_TYPE") or col.get("data_type") or "").lower(), + "is_nullable": col.get("IS_NULLABLE") or col.get("is_nullable"), + "column_key": col.get("COLUMN_KEY") or col.get("column_key") or "", + } + + for table, table_def in spec["tables"].items(): + live_cols = live_by_table.get(table) + if not live_cols: + violations.append(f"{table}: table missing from live source (contract v{spec.get('version')})") + continue + + contract_names = {c["name"] for c in table_def["columns"]} + live_names = set(live_cols.keys()) + + for extra in sorted(live_names - contract_names): + violations.append( + f"{table}.{extra}: unexpected column in live source (not in source contract)" + ) + + for col_spec in table_def["columns"]: + name = col_spec["name"] + if name not in live_cols: + violations.append(f"{table}.{name}: column missing from live source") + continue + + live = live_cols[name] + expected_type = col_spec["data_type"].lower() + if live["data_type"] != expected_type: + violations.append( + f"{table}.{name}: type mismatch — contract={expected_type}, live={live['data_type']}" + ) + + expected_null = "YES" if col_spec["nullable"] else "NO" + if live["is_nullable"] != expected_null: + violations.append( + f"{table}.{name}: nullability mismatch — contract={expected_null}, live={live['is_nullable']}" + ) + + expected_key = col_spec.get("column_key", "") + # Only enforce PRI/UNI from contract; MUL varies with source indexes. + if expected_key in ("PRI", "UNI") and live["column_key"] != expected_key: + violations.append( + f"{table}.{name}: key mismatch — contract={expected_key}, live={live['column_key']}" + ) + + live_fingerprint = fingerprint_from_mysql(info_schema_columns) + expected_fingerprint = load_expected_fingerprint() + violations.extend(compare_fingerprints(expected_fingerprint, live_fingerprint)) + + return violations + + +def compare_fingerprints( + expected: dict[str, str], live: dict[str, str] +) -> list[str]: + violations: list[str] = [] + for table in sorted(set(expected) | set(live)): + if table not in expected: + violations.append(f"{table}: unexpected table in live source") + elif table not in live: + violations.append(f"{table}: table removed from live source") + elif expected[table] != live[table]: + violations.append( + f"{table}: incompatible schema change (live source does not match contract)" + ) + return violations + + +def write_failure_artifact(messages: list[str]) -> None: + FAILED_SCHEMA_PATH.parent.mkdir(parents=True, exist_ok=True) + body = ( + "# Schema validation failed — ingestion stopped\n\n" + "Live source schema does not match the source-team contract.\n\n" + + "\n".join(f"- {m}" for m in messages) + ) + FAILED_SCHEMA_PATH.write_text(body, encoding="utf-8") + + +def validate_schema( + *, + info_schema_columns: list[dict[str, Any]] | None = None, + current_fingerprint: dict[str, str] | None = None, + actual_by_table: dict[str, set[str]] | None = None, +) -> list[str]: + """ + Full validation before ingestion. + + MySQL path: pass info_schema_columns (validates against source contract every time). + DuckDB/test path: pass current_fingerprint + actual_by_table. + """ + if info_schema_columns is not None: + return validate_live_against_source_contract(info_schema_columns) + + if current_fingerprint is None or actual_by_table is None: + raise ValueError( + "Provide info_schema_columns for MySQL, or current_fingerprint + actual_by_table" + ) + + violations = check_contract_columns(actual_by_table) + expected = load_expected_fingerprint() + violations.extend(compare_fingerprints(expected, current_fingerprint)) + return violations + + +def save_expected_fingerprint(fingerprint: dict[str, str]) -> None: + CONTRACTS_DIR.mkdir(parents=True, exist_ok=True) + SCHEMA_FINGERPRINT_PATH.write_text( + json.dumps(fingerprint, indent=2, sort_keys=True), encoding="utf-8" + ) + + +def export_contract_file() -> None: + """Sync column-list contract from spec (for catalog / legacy checks).""" + spec = load_source_contract_spec() + column_list = { + table: [c["name"] for c in table_def["columns"]] + for table, table_def in spec["tables"].items() + } + CONTRACTS_DIR.mkdir(parents=True, exist_ok=True) + SCHEMA_CONTRACT_PATH.write_text( + json.dumps(column_list, indent=2, sort_keys=True), encoding="utf-8" + ) diff --git a/submission/cdc-lakehouse/pipeline/warehouse_apply.py b/submission/cdc-lakehouse/pipeline/warehouse_apply.py new file mode 100644 index 0000000..af207a7 --- /dev/null +++ b/submission/cdc-lakehouse/pipeline/warehouse_apply.py @@ -0,0 +1,170 @@ +"""DuckDB warehouse current-state tables driven by CDC events.""" + +from __future__ import annotations + +import json +from datetime import datetime +from typing import Any + +import duckdb + +from pipeline.cdc_event import CDCEvent + +TABLE_MAP: dict[str, str] = { + "customers": "wh_customers", + "products": "wh_products", + "orders": "wh_orders", + "order_items": "wh_order_items", + "payments": "wh_payments", + "order_status_events": "wh_order_status_events", +} + +PK_MAP: dict[str, str] = { + "customers": "customer_id", + "products": "product_id", + "orders": "order_id", + "order_items": "order_item_id", + "payments": "payment_id", + "order_status_events": "status_event_id", +} + + +def create_warehouse_tables(conn: duckdb.DuckDBPyConnection) -> None: + conn.execute(""" + CREATE TABLE IF NOT EXISTS wh_customers ( + customer_id BIGINT PRIMARY KEY, + name VARCHAR, email VARCHAR, phone VARCHAR, status VARCHAR, + created_at TIMESTAMP, updated_at TIMESTAMP, is_deleted BOOLEAN, + _cdc_event_id BIGINT NOT NULL, _deleted BOOLEAN NOT NULL DEFAULT false + ) + """) + conn.execute(""" + CREATE TABLE IF NOT EXISTS wh_products ( + product_id BIGINT PRIMARY KEY, + sku VARCHAR, name VARCHAR, unit_price DECIMAL(12,2), + created_at TIMESTAMP, updated_at TIMESTAMP, is_deleted BOOLEAN, + _cdc_event_id BIGINT NOT NULL, _deleted BOOLEAN NOT NULL DEFAULT false + ) + """) + conn.execute(""" + CREATE TABLE IF NOT EXISTS wh_orders ( + order_id BIGINT PRIMARY KEY, + customer_id BIGINT, order_status VARCHAR, order_total DECIMAL(12,2), + created_at TIMESTAMP, updated_at TIMESTAMP, paid_at TIMESTAMP, + cancel_reason VARCHAR, is_deleted BOOLEAN, + _cdc_event_id BIGINT NOT NULL, _deleted BOOLEAN NOT NULL DEFAULT false + ) + """) + conn.execute(""" + CREATE TABLE IF NOT EXISTS wh_order_items ( + order_item_id BIGINT PRIMARY KEY, + order_id BIGINT, product_id BIGINT, quantity INTEGER, + unit_price DECIMAL(12,2), line_total DECIMAL(12,2), is_deleted BOOLEAN, + _cdc_event_id BIGINT NOT NULL, _deleted BOOLEAN NOT NULL DEFAULT false + ) + """) + conn.execute(""" + CREATE TABLE IF NOT EXISTS wh_payments ( + payment_id BIGINT PRIMARY KEY, + order_id BIGINT, payment_amount DECIMAL(12,2), payment_status VARCHAR, + created_at TIMESTAMP, updated_at TIMESTAMP, is_deleted BOOLEAN, + _cdc_event_id BIGINT NOT NULL, _deleted BOOLEAN NOT NULL DEFAULT false + ) + """) + conn.execute(""" + CREATE TABLE IF NOT EXISTS wh_order_status_events ( + status_event_id BIGINT PRIMARY KEY, + order_id BIGINT, from_status VARCHAR, to_status VARCHAR, + event_at TIMESTAMP, is_deleted BOOLEAN, + _cdc_event_id BIGINT NOT NULL, _deleted BOOLEAN NOT NULL DEFAULT false + ) + """) + + +def truncate_warehouse_tables(conn: duckdb.DuckDBPyConnection) -> None: + for wh_table in TABLE_MAP.values(): + conn.execute(f"DELETE FROM {wh_table}") + + +def apply_events(conn: duckdb.DuckDBPyConnection, events: list[CDCEvent]) -> None: + for event in sorted(events, key=lambda e: e.event_id): + _apply_one(conn, event) + + +def _apply_one(conn: duckdb.DuckDBPyConnection, event: CDCEvent) -> None: + wh_table = TABLE_MAP.get(event.source_table) + pk_col = PK_MAP.get(event.source_table) + if not wh_table or not pk_col: + return + + pk_val = event.primary_key[pk_col] + + if event.op == "D": + conn.execute( + f""" + UPDATE {wh_table} + SET _deleted = true, _cdc_event_id = ? + WHERE {pk_col} = ? AND (_cdc_event_id IS NULL OR _cdc_event_id < ?) + """, + [event.event_id, pk_val, event.event_id], + ) + existing = conn.execute( + f"SELECT COUNT(*) FROM {wh_table} WHERE {pk_col} = ?", [pk_val] + ).fetchone()[0] + if existing == 0: + row = {pk_col: pk_val, "_cdc_event_id": event.event_id, "_deleted": True} + _insert_row(conn, wh_table, row) + return + + data = _normalize_row(event.row) + data[pk_col] = pk_val + data["_cdc_event_id"] = event.event_id + data["_deleted"] = False + + current_seq = conn.execute( + f"SELECT _cdc_event_id FROM {wh_table} WHERE {pk_col} = ?", [pk_val] + ).fetchone() + if current_seq and current_seq[0] is not None and current_seq[0] >= event.event_id: + return + + if conn.execute( + f"SELECT COUNT(*) FROM {wh_table} WHERE {pk_col} = ?", [pk_val] + ).fetchone()[0]: + _update_row(conn, wh_table, pk_col, data) + else: + _insert_row(conn, wh_table, data) + + +def _normalize_row(row: dict[str, Any]) -> dict[str, Any]: + out: dict[str, Any] = {} + for key, value in row.items(): + if isinstance(value, datetime): + out[key] = value + elif isinstance(value, (dict, list)): + out[key] = json.dumps(value) + else: + out[key] = value + return out + + +def _insert_row( + conn: duckdb.DuckDBPyConnection, table: str, data: dict[str, Any] +) -> None: + cols = list(data.keys()) + placeholders = ", ".join(["?"] * len(cols)) + conn.execute( + f"INSERT INTO {table} ({', '.join(cols)}) VALUES ({placeholders})", + [data[c] for c in cols], + ) + + +def _update_row( + conn: duckdb.DuckDBPyConnection, table: str, pk_col: str, data: dict[str, Any] +) -> None: + pk_val = data[pk_col] + set_cols = [c for c in data if c != pk_col] + set_clause = ", ".join(f"{c} = ?" for c in set_cols) + conn.execute( + f"UPDATE {table} SET {set_clause} WHERE {pk_col} = ?", + [data[c] for c in set_cols] + [pk_val], + ) diff --git a/submission/cdc-lakehouse/scripts/check_schema_contracts.py b/submission/cdc-lakehouse/scripts/check_schema_contracts.py new file mode 100644 index 0000000..dd89e39 --- /dev/null +++ b/submission/cdc-lakehouse/scripts/check_schema_contracts.py @@ -0,0 +1,43 @@ +#!/usr/bin/env python3 +""" +Validate that schemas satisfy the source-team contract (CI-friendly DuckDB mirror). + +Exit 0 — contract satisfied. +Exit 1 — violations found. +""" + +from __future__ import annotations + +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +import duckdb + +from pipeline.schema_check import check_contract_columns +from source.models import SCHEMA_CONTRACT, create_source_tables + + +def main() -> int: + conn = duckdb.connect(":memory:") + create_source_tables(conn) + + actual = { + table: {row[0] for row in conn.execute(f"DESCRIBE {table}").fetchall()} + for table in SCHEMA_CONTRACT + } + violations = check_contract_columns(actual) + + if violations: + print("Schema contract violations:") + for v in violations: + print(f" ✗ {v}") + return 1 + + print(f"All schema contracts passed ({len(SCHEMA_CONTRACT)} tables checked).") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/submission/cdc-lakehouse/scripts/compile_schema_fingerprint.py b/submission/cdc-lakehouse/scripts/compile_schema_fingerprint.py new file mode 100644 index 0000000..43973d9 --- /dev/null +++ b/submission/cdc-lakehouse/scripts/compile_schema_fingerprint.py @@ -0,0 +1,34 @@ +#!/usr/bin/env python3 +""" +Compile expected schema_fingerprint.json FROM the source-team contract. + +Run when the source team publishes a new schema_contract_spec.json — not from live DB. +Data engineering does not bootstrap truth from production; the contract is authoritative. +""" + +from __future__ import annotations + +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from pipeline.schema_check import ( + expected_fingerprint_from_contract, + export_contract_file, + save_expected_fingerprint, +) + + +def main() -> int: + fp = expected_fingerprint_from_contract() + save_expected_fingerprint(fp) + export_contract_file() + print( + f"Compiled expected fingerprint for {len(fp)} tables from schema_contract_spec.json." + ) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/submission/cdc-lakehouse/scripts/restore_warehouse.py b/submission/cdc-lakehouse/scripts/restore_warehouse.py new file mode 100644 index 0000000..2bed1e6 --- /dev/null +++ b/submission/cdc-lakehouse/scripts/restore_warehouse.py @@ -0,0 +1,67 @@ +#!/usr/bin/env python3 +"""Rebuild warehouse snapshot from lake Parquet events as-of a timestamp.""" + +from __future__ import annotations + +import argparse +import os +import sys +from datetime import datetime, timezone + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +import duckdb + +from pipeline.config import LAKE_DIR, WAREHOUSE_PATH +from pipeline.lake_writer import read_all_lake_events +from pipeline.warehouse_apply import ( + apply_events, + create_warehouse_tables, + truncate_warehouse_tables, +) + + +def _parse_ts(value: str) -> datetime: + dt = datetime.fromisoformat(value.replace("Z", "+00:00")) + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return dt + + +def restore(as_of: datetime, lake_dir=None) -> int: + events = read_all_lake_events(lake_dir or LAKE_DIR) + filtered = [ + e + for e in events + if ( + e.event_ts.replace(tzinfo=timezone.utc) + if e.event_ts.tzinfo is None + else e.event_ts.astimezone(timezone.utc) + ) + <= as_of.astimezone(timezone.utc) + ] + + WAREHOUSE_PATH.parent.mkdir(parents=True, exist_ok=True) + conn = duckdb.connect(str(WAREHOUSE_PATH)) + create_warehouse_tables(conn) + truncate_warehouse_tables(conn) + apply_events(conn, filtered) + conn.close() + + print( + f"Restored warehouse from {len(filtered)} lake events as-of {as_of.isoformat()}." + ) + return 0 + + +def main() -> int: + parser = argparse.ArgumentParser( + description="Point-in-time warehouse restore from lake" + ) + parser.add_argument("--as-of", required=True, help="ISO-8601 timestamp (UTC)") + args = parser.parse_args() + return restore(_parse_ts(args.as_of)) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/submission/cdc-lakehouse/scripts/run_data_quality_checks.py b/submission/cdc-lakehouse/scripts/run_data_quality_checks.py new file mode 100644 index 0000000..3ee7458 --- /dev/null +++ b/submission/cdc-lakehouse/scripts/run_data_quality_checks.py @@ -0,0 +1,123 @@ +#!/usr/bin/env python3 +"""Run warehouse data quality checks (system + business parity).""" + +from __future__ import annotations + +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +import duckdb + +from pipeline.config import WAREHOUSE_PATH +from pipeline.warehouse_apply import create_warehouse_tables + + +def _check(name: str, sql: str, conn: duckdb.DuckDBPyConnection) -> list[str]: + count = conn.execute(sql).fetchone()[0] + if count: + return [f"{name}: {count} violation(s)"] + return [] + + +def run_checks(conn: duckdb.DuckDBPyConnection) -> list[str]: + violations: list[str] = [] + + # System: PK uniqueness (implicit via PK, but check active rows) + violations.extend( + _check( + "duplicate_active_customers", + "SELECT COUNT(*) - COUNT(DISTINCT customer_id) FROM wh_customers WHERE NOT _deleted", + conn, + ) + ) + + # FK: order_items → orders + violations.extend( + _check( + "orphan_order_items", + """ + SELECT COUNT(*) FROM wh_order_items oi + LEFT JOIN wh_orders o ON oi.order_id = o.order_id AND NOT o._deleted + WHERE NOT oi._deleted AND o.order_id IS NULL + """, + conn, + ) + ) + + # Business: non-negative money + violations.extend( + _check( + "negative_order_totals", + "SELECT COUNT(*) FROM wh_orders WHERE NOT _deleted AND order_total < 0", + conn, + ) + ) + violations.extend( + _check( + "negative_line_totals", + "SELECT COUNT(*) FROM wh_order_items WHERE NOT _deleted AND line_total < 0", + conn, + ) + ) + + # Business: order_total vs line items (stored total must match sum of lines) + violations.extend( + _check( + "order_total_mismatch", + """ + SELECT COUNT(*) FROM wh_orders o + JOIN ( + SELECT order_id, SUM(line_total) AS items_total + FROM wh_order_items + WHERE NOT _deleted + GROUP BY order_id + ) x ON o.order_id = x.order_id + WHERE NOT o._deleted AND ABS(o.order_total - x.items_total) > 0.01 + """, + conn, + ) + ) + + # Business: paid_at after created_at + violations.extend( + _check( + "paid_before_created", + """ + SELECT COUNT(*) FROM wh_orders + WHERE NOT _deleted AND paid_at IS NOT NULL AND paid_at < created_at + """, + conn, + ) + ) + + return violations + + +def main() -> int: + if not WAREHOUSE_PATH.exists(): + # CI: create empty warehouse tables so script exits cleanly + WAREHOUSE_PATH.parent.mkdir(parents=True, exist_ok=True) + conn = duckdb.connect(str(WAREHOUSE_PATH)) + create_warehouse_tables(conn) + conn.close() + print("Warehouse not populated yet; no violations in empty tables.") + return 0 + + conn = duckdb.connect(str(WAREHOUSE_PATH)) + violations = run_checks(conn) + conn.close() + + if violations: + print("Data quality violations:") + for v in violations: + print(f" ✗ {v}") + return 1 + + print("All data quality checks passed.") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/submission/cdc-lakehouse/scripts/run_ingestion.sh b/submission/cdc-lakehouse/scripts/run_ingestion.sh new file mode 100755 index 0000000..606e639 --- /dev/null +++ b/submission/cdc-lakehouse/scripts/run_ingestion.sh @@ -0,0 +1,5 @@ +#!/usr/bin/env bash +set -euo pipefail +ROOT="$(cd "$(dirname "$0")/.." && pwd)" +cd "$ROOT" +python -m pipeline.ingest "$@" diff --git a/submission/cdc-lakehouse/scripts/validate_catalog.py b/submission/cdc-lakehouse/scripts/validate_catalog.py new file mode 100644 index 0000000..8e6284d --- /dev/null +++ b/submission/cdc-lakehouse/scripts/validate_catalog.py @@ -0,0 +1,60 @@ +#!/usr/bin/env python3 +"""Validate catalog metadata file.""" + +from __future__ import annotations + +import json +import os +import sys +from pathlib import Path + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +ROOT = Path(__file__).resolve().parents[1] +CATALOG_PATH = ROOT / "catalog" / "catalog.json" + + +def main() -> int: + if not CATALOG_PATH.exists(): + print(f"Missing catalog file: {CATALOG_PATH}") + return 1 + + data = json.loads(CATALOG_PATH.read_text(encoding="utf-8")) + datasets = data.get("datasets", []) + if not datasets: + print("Catalog has no datasets.") + return 1 + + layers = {d.get("layer") for d in datasets} + if "lake" not in layers or "warehouse" not in layers: + print("Catalog must include both lake and warehouse datasets.") + return 1 + + required = { + "name", + "layer", + "description", + "owner", + "consumers", + "update_cadence", + "location", + } + violations: list[str] = [] + for ds in datasets: + missing = required - set(ds.keys()) + if missing: + violations.append( + f"{ds.get('name', '?')}: missing fields {sorted(missing)}" + ) + + if violations: + for v in violations: + print(f" ✗ {v}") + return 1 + + print(f"Catalog valid ({len(datasets)} datasets).") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) From fd12220cc8d89fd9229476c7df32890ec533f5db Mon Sep 17 00:00:00 2001 From: Shravani More Date: Wed, 27 May 2026 18:05:11 +0530 Subject: [PATCH 4/4] test: add validation tests, conftest, and catalog mapping --- submission/cdc-lakehouse/catalog/catalog.json | 67 +++++++++++ submission/cdc-lakehouse/conftest.py | 6 + submission/cdc-lakehouse/tests/conftest.py | 15 +++ .../cdc-lakehouse/tests/test_catalog.py | 12 ++ .../cdc-lakehouse/tests/test_cdc_event.py | 30 +++++ .../cdc-lakehouse/tests/test_checkpoint.py | 10 ++ submission/cdc-lakehouse/tests/test_lake.py | 39 ++++++ .../cdc-lakehouse/tests/test_restore.py | 58 +++++++++ .../tests/test_schema_contracts.py | 88 ++++++++++++++ .../cdc-lakehouse/tests/test_warehouse.py | 112 ++++++++++++++++++ 10 files changed, 437 insertions(+) create mode 100644 submission/cdc-lakehouse/catalog/catalog.json create mode 100644 submission/cdc-lakehouse/conftest.py create mode 100644 submission/cdc-lakehouse/tests/conftest.py create mode 100644 submission/cdc-lakehouse/tests/test_catalog.py create mode 100644 submission/cdc-lakehouse/tests/test_cdc_event.py create mode 100644 submission/cdc-lakehouse/tests/test_checkpoint.py create mode 100644 submission/cdc-lakehouse/tests/test_lake.py create mode 100644 submission/cdc-lakehouse/tests/test_restore.py create mode 100644 submission/cdc-lakehouse/tests/test_schema_contracts.py create mode 100644 submission/cdc-lakehouse/tests/test_warehouse.py diff --git a/submission/cdc-lakehouse/catalog/catalog.json b/submission/cdc-lakehouse/catalog/catalog.json new file mode 100644 index 0000000..a192ac4 --- /dev/null +++ b/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" + } + } + ] +} diff --git a/submission/cdc-lakehouse/conftest.py b/submission/cdc-lakehouse/conftest.py new file mode 100644 index 0000000..5334305 --- /dev/null +++ b/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)) diff --git a/submission/cdc-lakehouse/tests/conftest.py b/submission/cdc-lakehouse/tests/conftest.py new file mode 100644 index 0000000..6ebcb9a --- /dev/null +++ b/submission/cdc-lakehouse/tests/conftest.py @@ -0,0 +1,15 @@ +import pytest + + +@pytest.fixture +def sample_customer_row() -> dict: + return { + "customer_id": 1, + "name": "Alice", + "email": "alice@example.com", + "phone": None, + "status": "ACTIVE", + "created_at": "2026-05-26T10:00:00", + "updated_at": "2026-05-26T10:00:00", + "is_deleted": 0, + } diff --git a/submission/cdc-lakehouse/tests/test_catalog.py b/submission/cdc-lakehouse/tests/test_catalog.py new file mode 100644 index 0000000..1190c08 --- /dev/null +++ b/submission/cdc-lakehouse/tests/test_catalog.py @@ -0,0 +1,12 @@ +"""Catalog metadata tests.""" + +import json +from pathlib import Path + + +def test_catalog_has_lake_and_warehouse(): + catalog_path = Path(__file__).resolve().parents[1] / "catalog" / "catalog.json" + data = json.loads(catalog_path.read_text(encoding="utf-8")) + layers = {d["layer"] for d in data["datasets"]} + assert "lake" in layers + assert "warehouse" in layers diff --git a/submission/cdc-lakehouse/tests/test_cdc_event.py b/submission/cdc-lakehouse/tests/test_cdc_event.py new file mode 100644 index 0000000..5fa71b4 --- /dev/null +++ b/submission/cdc-lakehouse/tests/test_cdc_event.py @@ -0,0 +1,30 @@ +"""CDC event model tests.""" + +from datetime import datetime, timezone + +import pytest + +from pipeline.cdc_event import CDCEvent + + +def test_valid_ops(): + CDCEvent( + event_id=1, + source_table="customers", + op="I", + pk_json='{"customer_id": 1}', + row_json='{"customer_id": 1}', + event_ts=datetime.now(timezone.utc), + ) + + +def test_invalid_op_raises(): + with pytest.raises(ValueError, match="Invalid op"): + CDCEvent( + event_id=1, + source_table="customers", + op="X", + pk_json="{}", + row_json="{}", + event_ts=datetime.now(timezone.utc), + ) diff --git a/submission/cdc-lakehouse/tests/test_checkpoint.py b/submission/cdc-lakehouse/tests/test_checkpoint.py new file mode 100644 index 0000000..075b496 --- /dev/null +++ b/submission/cdc-lakehouse/tests/test_checkpoint.py @@ -0,0 +1,10 @@ +"""Checkpoint persistence tests.""" + +from pipeline.checkpoint import load_checkpoint, save_checkpoint + + +def test_checkpoint_roundtrip(tmp_path): + cp = tmp_path / "checkpoint.json" + assert load_checkpoint(cp) == 0 + save_checkpoint(42, cp) + assert load_checkpoint(cp) == 42 diff --git a/submission/cdc-lakehouse/tests/test_lake.py b/submission/cdc-lakehouse/tests/test_lake.py new file mode 100644 index 0000000..329bacd --- /dev/null +++ b/submission/cdc-lakehouse/tests/test_lake.py @@ -0,0 +1,39 @@ +"""Lake Parquet writer tests.""" + +from datetime import datetime, timezone + +from pipeline.cdc_event import CDCEvent +from pipeline.lake_writer import read_all_lake_events, write_events_to_lake + + +def test_write_and_read_lake_roundtrip(tmp_path, monkeypatch): + monkeypatch.setenv("LAKE_DIR", str(tmp_path / "lake")) + from pipeline import config + + config.LAKE_DIR = tmp_path / "lake" + + events = [ + CDCEvent( + event_id=1, + source_table="customers", + op="I", + pk_json='{"customer_id": 1}', + row_json='{"customer_id": 1, "name": "Alice"}', + event_ts=datetime(2026, 5, 26, 12, 0, tzinfo=timezone.utc), + ), + CDCEvent( + event_id=2, + source_table="orders", + op="U", + pk_json='{"order_id": 10}', + row_json='{"order_id": 10, "order_status": "PAID"}', + event_ts=datetime(2026, 5, 26, 12, 1, tzinfo=timezone.utc), + ), + ] + paths = write_events_to_lake(events, run_id="testrun") + assert len(paths) >= 1 + + loaded = read_all_lake_events(tmp_path / "lake") + assert len(loaded) == 2 + assert loaded[0].event_id == 1 + assert loaded[1].op == "U" diff --git a/submission/cdc-lakehouse/tests/test_restore.py b/submission/cdc-lakehouse/tests/test_restore.py new file mode 100644 index 0000000..e4732df --- /dev/null +++ b/submission/cdc-lakehouse/tests/test_restore.py @@ -0,0 +1,58 @@ +"""Point-in-time restore from lake.""" + +from datetime import datetime, timezone + +import duckdb + +from pipeline.cdc_event import CDCEvent +from pipeline.lake_writer import write_events_to_lake +from pipeline.warehouse_apply import ( + apply_events, + create_warehouse_tables, + truncate_warehouse_tables, +) + + +def test_restore_as_of_filters_events(tmp_path, monkeypatch): + monkeypatch.setenv("LAKE_DIR", str(tmp_path / "lake")) + from pipeline import config + + config.LAKE_DIR = tmp_path / "lake" + wh_path = tmp_path / "wh.duckdb" + + t1 = datetime(2026, 5, 26, 10, 0, tzinfo=timezone.utc) + t2 = datetime(2026, 5, 26, 14, 0, tzinfo=timezone.utc) + + events = [ + CDCEvent( + 1, + "customers", + "I", + '{"customer_id": 1}', + '{"customer_id": 1, "name": "V1", "email": null, "phone": null, "status": "ACTIVE", "created_at": "2026-05-26T10:00:00", "updated_at": "2026-05-26T10:00:00", "is_deleted": 0}', + t1, + ), + CDCEvent( + 2, + "customers", + "U", + '{"customer_id": 1}', + '{"customer_id": 1, "name": "V2", "email": null, "phone": null, "status": "ACTIVE", "created_at": "2026-05-26T10:00:00", "updated_at": "2026-05-26T14:00:00", "is_deleted": 0}', + t2, + ), + ] + write_events_to_lake(events, run_id="r1") + + from pipeline.lake_writer import read_all_lake_events + + as_of = datetime(2026, 5, 26, 12, 0, tzinfo=timezone.utc) + filtered = [e for e in read_all_lake_events() if e.event_ts <= as_of] + + conn = duckdb.connect(str(wh_path)) + create_warehouse_tables(conn) + truncate_warehouse_tables(conn) + apply_events(conn, filtered) + name = conn.execute( + "SELECT name FROM wh_customers WHERE customer_id = 1" + ).fetchone()[0] + assert name == "V1" diff --git a/submission/cdc-lakehouse/tests/test_schema_contracts.py b/submission/cdc-lakehouse/tests/test_schema_contracts.py new file mode 100644 index 0000000..640b552 --- /dev/null +++ b/submission/cdc-lakehouse/tests/test_schema_contracts.py @@ -0,0 +1,88 @@ +"""Schema contract and fingerprint tests.""" + +import duckdb + +from pipeline.schema_check import ( + check_contract_columns, + compare_fingerprints, + contract_spec_to_information_schema_rows, + load_source_contract_spec, + validate_live_against_source_contract, +) +from source.models import SCHEMA_CONTRACT, create_source_tables + + +def test_fresh_source_passes_contracts(): + conn = duckdb.connect(":memory:") + create_source_tables(conn) + actual = { + t: {row[0] for row in conn.execute(f"DESCRIBE {t}").fetchall()} + for t in SCHEMA_CONTRACT + } + assert check_contract_columns(actual) == [] + + +def test_dropped_column_detected(): + conn = duckdb.connect(":memory:") + create_source_tables(conn) + conn.execute("ALTER TABLE customers DROP COLUMN email") + actual = { + t: {row[0] for row in conn.execute(f"DESCRIBE {t}").fetchall()} + for t in SCHEMA_CONTRACT + } + violations = check_contract_columns(actual) + assert any("email" in v for v in violations) + + +def test_fingerprint_drift_detected(): + baseline = {"customers": "abc123"} + current = {"customers": "different"} + assert compare_fingerprints(baseline, current) + + +def test_validate_schema_stops_on_drift(): + spec = load_source_contract_spec() + rows = contract_spec_to_information_schema_rows(spec) + # Simulate source dropping a column + rows = [r for r in rows if r["COLUMN_NAME"] != "email"] + violations = validate_live_against_source_contract(rows) + assert any("email" in v or "missing" in v for v in violations) + + +def test_live_matching_contract_passes(): + spec = load_source_contract_spec() + rows = contract_spec_to_information_schema_rows(spec) + assert validate_live_against_source_contract(rows) == [] + + +def test_pipeline_stops_on_contract_violation(): + from pipeline.cdc_event import CDCEvent + + conn = duckdb.connect(":memory:") + conn.execute(""" + CREATE TABLE customers ( + customer_id BIGINT PRIMARY KEY, + name VARCHAR NOT NULL, + status VARCHAR NOT NULL, + created_at TIMESTAMP NOT NULL, + updated_at TIMESTAMP NOT NULL, + is_deleted BOOLEAN NOT NULL DEFAULT false + ) + """) + actual = { + "customers": {row[0] for row in conn.execute("DESCRIBE customers").fetchall()} + } + violations = check_contract_columns(actual) + events: list[CDCEvent] = [] + if not violations: + events.append( + CDCEvent( + 1, + "customers", + "I", + "{}", + "{}", + __import__("datetime").datetime.utcnow(), + ) + ) + assert len(events) == 0 diff --git a/submission/cdc-lakehouse/tests/test_warehouse.py b/submission/cdc-lakehouse/tests/test_warehouse.py new file mode 100644 index 0000000..10c3354 --- /dev/null +++ b/submission/cdc-lakehouse/tests/test_warehouse.py @@ -0,0 +1,112 @@ +"""Warehouse apply logic tests.""" + +from datetime import datetime, timezone + +import duckdb + +from pipeline.cdc_event import CDCEvent +from pipeline.warehouse_apply import apply_events, create_warehouse_tables + + +def _event(eid: int, table: str, op: str, pk: int, row: dict) -> CDCEvent: + import json + + pk_name = { + "customers": "customer_id", + "orders": "order_id", + }[table] + return CDCEvent( + event_id=eid, + source_table=table, + op=op, + pk_json=json.dumps({pk_name: pk}), + row_json=json.dumps(row), + event_ts=datetime(2026, 5, 26, tzinfo=timezone.utc), + ) + + +def test_insert_update_delete_lifecycle(): + conn = duckdb.connect(":memory:") + create_warehouse_tables(conn) + + apply_events( + conn, + [ + _event( + 1, + "customers", + "I", + 1, + { + "customer_id": 1, + "name": "Alice", + "email": "a@x.com", + "phone": None, + "status": "ACTIVE", + "created_at": "2026-05-26T10:00:00", + "updated_at": "2026-05-26T10:00:00", + "is_deleted": 0, + }, + ), + _event( + 2, + "customers", + "U", + 1, + { + "customer_id": 1, + "name": "Alice Updated", + "email": "a@x.com", + "phone": None, + "status": "SUSPENDED", + "created_at": "2026-05-26T10:00:00", + "updated_at": "2026-05-26T11:00:00", + "is_deleted": 0, + }, + ), + _event( + 3, + "customers", + "D", + 1, + {"customer_id": 1}, + ), + ], + ) + + row = conn.execute( + "SELECT name, status, _deleted, _cdc_event_id FROM wh_customers WHERE customer_id = 1" + ).fetchone() + assert row[2] is True + assert row[3] == 3 + + +def test_idempotent_replay_skips_stale_events(): + conn = duckdb.connect(":memory:") + create_warehouse_tables(conn) + + insert = _event( + 10, + "customers", + "I", + 5, + { + "customer_id": 5, + "name": "Bob", + "email": None, + "phone": None, + "status": "ACTIVE", + "created_at": "2026-05-26T10:00:00", + "updated_at": "2026-05-26T10:00:00", + "is_deleted": 0, + }, + ) + apply_events(conn, [insert]) + apply_events(conn, [insert]) # duplicate replay + + count = conn.execute("SELECT COUNT(*) FROM wh_customers").fetchone()[0] + assert count == 1 + name = conn.execute( + "SELECT name FROM wh_customers WHERE customer_id = 5" + ).fetchone()[0] + assert name == "Bob"