diff --git a/submission/ecommerce-cdc/.gitignore b/submission/ecommerce-cdc/.gitignore new file mode 100644 index 0000000..37a528c --- /dev/null +++ b/submission/ecommerce-cdc/.gitignore @@ -0,0 +1,6 @@ +__pycache__/ +*.pyc +.pytest_cache/ +.venv/ +*.duckdb +.data/ diff --git a/submission/ecommerce-cdc/PR_DESCRIPTION.md b/submission/ecommerce-cdc/PR_DESCRIPTION.md new file mode 100644 index 0000000..8e9e815 --- /dev/null +++ b/submission/ecommerce-cdc/PR_DESCRIPTION.md @@ -0,0 +1,70 @@ +# CDC Lakehouse Reliability — E-commerce + +A small but deeply correct CDC pipeline keeping a lake (full history) and a warehouse +(latest snapshot + time travel) in sync with an e-commerce source. +Stack: PostgreSQL (source design) · Python (ingestion) · DuckDB (lake + warehouse) · +pytest. CDC is simulated in-process (dependency-free); the production analogue is +documented throughout. Full design: `submission/ecommerce-cdc/SOLUTION_APPROACH.md`. + +## 1. Source schema design +- **Tables:** `customers`, `products`, `orders` (strong); `order_items`, `payments` (weak). +- **Strong vs weak:** strong entities have independent identity; `order_items` and + `payments` have no meaning without their parent `order` (identity completed by the FK). +- **Keys/relationships:** PKs on every table; FKs `orders→customers`, + `order_items→orders/products`, `payments→orders` (1—N throughout). +- **Indexes:** FK columns + `updated_at` (the change-capture path). +- **Types:** decimals (`price/total/amount/line_total`), timestamps, enums + (`status/method`), nullable (`settled_at`). +- **Validation rules:** `total = Σ line_total`, `line_total = qty*unit_price`, + non-negative money, `price>0`, `quantity>0`, `settled_at ≥ created_at`, enum domains, + legal status transitions. + +## 2. CDC strategy +- Every mutation → a `CDCRecord(op, table, pk, before, after, captured_at, sequence)`. +- **`sequence`** is a monotonic offset (Kafka offset / Postgres LSN analogue) — the sole + ordering + checkpoint primitive. +- **Replay/restart:** checkpoint stores the last applied sequence; restart replays only + `records_since(checkpoint)`. +- **Duplicates:** lake PK + `ON CONFLICT DO NOTHING`; warehouse skips events whose + sequence ≤ the row's current `_cdc_seq` (also handles out-of-order). +- **Deletes:** captured as `delete` events; snapshot soft-deletes, SCD2 history closes + the version. + +## 3. Lake & warehouse modeling +- **Lake (`lake_cdc_events`):** append-only, idempotent, stores full row image as JSON — + every change is recoverable. +- **Warehouse current (`wh_*`):** upsert by PK + `_deleted` soft-delete = latest snapshot. +- **Warehouse history (`wh_*_history`):** SCD2 (`_valid_from_seq`/`_valid_to_seq`/ + `_is_current`). +- **Time travel:** `reconstruct_as_of(table, seq)` reads the version valid at `seq`. +- **Restore:** `restore_from_lake(target_seq)` deterministically rebuilds the snapshot by + replaying the lake. + +## 4. Schema change safety +- Declared `SCHEMA_CONTRACT` (columns, types, nullability, enum domains, keys). +- Each cycle diffs live source schema vs contract. **Breaking** (dropped/renamed column, + type change, nullable tightening, enum shrink) → raise `SchemaIncompatibilityError`, + **stop ingestion, write nothing**, exit non-zero. **Additive** changes warn and continue. +- CI gate: `scripts/check_schema_contracts.py`. + +## 5. Validation parity +- **System:** PK uniqueness, not-null, referential integrity, enum domains. +- **Business:** total = Σ line_total, line_total = qty*unit_price, non-negative money, + `price>0`, `quantity>0`, `settled_at ≥ created_at`, legal status transitions (validated + from ordered lake history). +- Surfaced via `validation/checks.py`, `scripts/run_data_quality_checks.py`, and tests. + +## 6. Catalog exposure +- `catalog/catalog.json` registers every lake + warehouse dataset (layer, owner, + consumers, cadence, query path, per-column schema). Gate: `scripts/validate_catalog.py`; + a test also asserts catalog ↔ live tables stay in sync. + +## 7. Responsible AI usage +AI assistance was used to scaffold boilerplate and draft documentation. All schema +design, CDC/SCD2 semantics, schema-safety rules, and validation logic were reviewed and +verified by running the test suite (`41 passed`) and the CI gate scripts locally. + +## Testing +`pytest -q` → **41 passed**, covering modeling/constraints, CDC correctness +(insert/update/delete/duplicate/replay/restart), schema-change safety, warehouse +snapshot + time travel + restore, and catalog. Red→Blue→Green discipline followed. diff --git a/submission/ecommerce-cdc/README.md b/submission/ecommerce-cdc/README.md new file mode 100644 index 0000000..658be80 --- /dev/null +++ b/submission/ecommerce-cdc/README.md @@ -0,0 +1,88 @@ +# E-commerce CDC Lakehouse + +A change-data-capture pipeline that keeps a **lake** (full change history) and a +**warehouse** (latest snapshot + time travel) in sync with an e-commerce source +system, with schema-drift safety and validation parity. + +> Design rationale and the full step-by-step approach live in +> [SOLUTION_APPROACH.md](./SOLUTION_APPROACH.md). + +## Stack + +- **PostgreSQL** — canonical source design (`source/postgres_ddl.sql`) +- **Python** — CDC capture, checkpointing, schema safety, orchestration +- **DuckDB** — lake + warehouse (object-storage / cloud-warehouse analogue) +- **pytest** — system + business validation + +CDC runs in two interchangeable modes: + +- **Simulated in-process** (default) — runs with no external services; used by the test + suite and CI. +- **Real PostgreSQL source via Docker** (optional) — insert/update/delete in a real + Postgres and capture the changes into the lake + warehouse. See + [TESTING_POSTGRES.md](./TESTING_POSTGRES.md). + +The production analogue (Debezium / Postgres WAL) is documented at every layer. + +## Layout + +``` +source/ Postgres DDL + DuckDB models + SCHEMA_CONTRACT +ingestion/ cdc capture · checkpoint · schema-contract safe-stop · pipeline · seed +lake/ append-only, idempotent CDC event history +warehouse/ current snapshot + SCD2 history + restore / time travel +validation/ system + business parity checks +catalog/ catalog.json dataset registry +postgres/ optional real-Postgres source bridge (snapshot-diff CDC via psycopg) +scripts/ CI gates (schema · data quality · catalog) + Postgres demo CLIs +tests/ red/blue/green tests across all categories +docker-compose.yml local Postgres that auto-loads the source schema +``` + +## Setup & run (simulated, no external services) + +```bash +cd submission/ecommerce-cdc +python -m venv .venv && . .venv/bin/activate +pip install -r requirements.txt +``` + +```bash +pytest -q # full test suite (41 tests) +python -m ingestion.pipeline # end-to-end demo (ingest + time travel + restore) + +python scripts/check_schema_contracts.py # schema safety gate +python scripts/run_data_quality_checks.py # validation parity gate +python scripts/validate_catalog.py # catalog gate +``` + +## Run against a real PostgreSQL source (optional, Docker) + +Requires Docker (e.g. Colima); the Postgres driver (`psycopg`) is already in +`requirements.txt`. Full walkthrough: [TESTING_POSTGRES.md](./TESTING_POSTGRES.md). + +```bash +docker-compose up -d # start Postgres (schema auto-created) + +python scripts/pg_seed.py # insert realistic sample rows into Postgres +python scripts/pg_sync.py # capture changes -> lake + warehouse + report +python scripts/pg_mutate.py # apply updates/deletes (legal transitions) +python scripts/pg_sync.py # re-sync to see update/delete capture +``` + +Postgres connection: `localhost:5432`, db `kulu_ecom`, user/pass `kulu`/`kulu`. +The persistent lakehouse lives at `.data/lakehouse.duckdb`. + +## What it demonstrates + +| Requirement | Where | +|---|---| +| Source model (strong/weak entities, keys, indexes, enums) | `source/` | +| CDC capture + checkpoint/replay/restart | `ingestion/cdc.py`, `ingestion/checkpoint.py` | +| Lake retains every change (idempotent) | `lake/lake.py` | +| Warehouse latest snapshot + SCD2 time travel + restore | `warehouse/warehouse.py` | +| Schema-change detection + fail-closed stop | `ingestion/schema_contract.py` | +| Validation parity (system + business) | `validation/checks.py` | +| Catalog exposure | `catalog/catalog.json` | +| Reliability (duplicates, deletes, out-of-order, restart) | see SOLUTION_APPROACH §10 | +| Real Postgres source + CDC capture (optional) | `postgres/`, `scripts/pg_*.py`, `docker-compose.yml` | diff --git a/submission/ecommerce-cdc/SOLUTION_APPROACH.md b/submission/ecommerce-cdc/SOLUTION_APPROACH.md new file mode 100644 index 0000000..bae702a --- /dev/null +++ b/submission/ecommerce-cdc/SOLUTION_APPROACH.md @@ -0,0 +1,249 @@ +# Solution Approach — E-commerce CDC Lakehouse + +A change-data-capture (CDC) pipeline that keeps two derived stores in sync with an +e-commerce source system: + +- a **lake** — append-only, immutable history of *every* change, and +- a **warehouse** — a current-state snapshot plus SCD2 history for point-in-time + reconstruction (time travel), + +with schema-drift safety, validation parity, and a dataset catalog. + +The pipeline runs in two interchangeable modes that share the *same* lake + warehouse + +validation code: + +- **Simulated, in-process** (default) — no external services; used by the test suite and + CI gates. +- **Real PostgreSQL source via Docker** (optional) — insert/update/delete in a real + Postgres and capture those changes (snapshot-diff CDC). See + [TESTING_POSTGRES.md](./TESTING_POSTGRES.md). + +The production analogue (Debezium / Postgres logical replication reading the WAL) is +documented at every layer; only the *capture* mechanism changes, never anything +downstream. + +--- + +## 1. Architecture overview + +``` + ┌──────────────┐ CDCRecord(seq) ┌──────────────────────────────┐ + source ──▶ │ CDC capture │ ─────────────────▶ │ lake (append-only history) │ + (Postgres │ cdc.py / │ └───────────────┬──────────────┘ + or DuckDB │ bridge.py) │ │ replay + sim) └──────┬───────┘ ▼ + │ checkpoint (last seq) ┌──────────────────────────────┐ + └────────────────────────────▶ │ warehouse │ + │ wh_* (snapshot) │ + │ wh_*_history (SCD2) │ + └───────────────┬──────────────┘ + ▼ + validation parity + catalog exposure +``` + +Every change becomes a `CDCRecord` carrying a monotonically increasing **`sequence`** +(the local analogue of a Kafka offset / Postgres LSN). `sequence` is the single ordering +and checkpoint primitive used by every layer. + +--- + +## 2. Source schema design (`source/`) + +| Table | Kind | PK | Foreign keys | +|---|---|---|---| +| `customers` | strong | `customer_id` | — | +| `products` | strong | `product_id` | — | +| `orders` | strong | `order_id` | `customer_id → customers` | +| `order_items` | weak | `order_item_id` | `order_id → orders`, `product_id → products` | +| `payments` | weak | `payment_id` | `order_id → orders` | + +- **Strong vs weak:** strong entities have independent identity; `order_items` and + `payments` have no meaning without their parent `order` (identity completed by the FK). +- **Types:** decimals for money (`price`, `total`, `amount`, `line_total`), timestamps, + enums for `status`/`method`, and a nullable `settled_at`. +- **Enums:** customer/product/order/payment status domains + payment method. +- **Indexes:** FK columns and `updated_at` (the change-capture path). +- **Business invariants:** `order.total = Σ line_total`, `line_total = quantity * + unit_price`, money ≥ 0, `price > 0`, `quantity > 0`, `settled_at ≥ created_at`, enum + domains, and legal status transitions only. +- **Expected change patterns:** `customers`/`products` are mostly inserts with occasional + status/price updates and rare deletes; `orders` are insert-then-update heavy (status + walks `pending → paid → shipped → delivered`); `order_items` are insert-once and rarely + change; `payments` are insert-then-update (`pending → settled`). The change-capture + path (`updated_at` + diff) is indexed accordingly. + +The contract lives once in `source/models.py` as `SCHEMA_CONTRACT` (a typed +`TableContract`/`ColumnSpec` structure). `source/postgres_ddl.sql` is the canonical +PostgreSQL DDL (ENUM types, PK/FK, CHECKs, indexes) that the DuckDB model mirrors, so the +same design holds whether running simulated or against real Postgres. + +--- + +## 3. CDC capture (`ingestion/cdc.py`, `postgres/bridge.py`) + +- A `CDCRecord` carries `operation` (insert/update/delete), `table`, `primary_key`, + `before`/`after` row images, `captured_at`, and `sequence`. +- **Simulated:** `CDCCapture` is an in-process change log that assigns sequences and + supports `records_since(offset)` for replay. +- **Real Postgres:** `bridge.sync()` reads the current state of every source table and + diffs it against the last-seen snapshot stored in DuckDB (`_pg_snapshot`); the + differences become insert/update/delete `CDCRecord`s. Sequencing continues + monotonically across syncs. +- Both modes resume sequencing from the persisted checkpoint + (`max(checkpoint, MAX(lake.sequence))`), so a crash between lake-write and checkpoint + cannot reuse a sequence. + +--- + +## 4. Checkpoint, replay & restart (`ingestion/checkpoint.py`) + +- The last successfully applied `sequence` per consumer is persisted in + `_ingest_checkpoint`. +- Each cycle processes only `records_since(checkpoint)` and advances the checkpoint + **only after** a successful lake write + warehouse apply. +- Restart after a crash resumes from the last checkpoint; re-running with no new changes + is a no-op. + +--- + +## 5. Lake — full history (`lake/lake.py`) + +- `lake_cdc_events` is **append-only**: every change is stored exactly once with its full + row image as JSON. +- **Idempotent:** `sequence` is the primary key and appends use `ON CONFLICT DO NOTHING`, + so duplicate/replayed events cannot corrupt history. +- It is the durable source of truth for replay and for rebuilding the warehouse. +- Production analogue: Parquet/Delta files on object storage, partitioned by table and + capture date. + +--- + +## 6. Warehouse — snapshot + time travel (`warehouse/warehouse.py`) + +For every source table the warehouse keeps two coordinated models derived from the same +sequence-ordered stream (so they can never silently drift): + +- **`wh_` (current snapshot):** upsert by PK; deletes set `_deleted = true` + (soft delete). `_cdc_seq` records the last applied sequence per row. +- **`wh_
_history` (SCD2):** `_valid_from_seq` / `_valid_to_seq` / `_is_current` + versions, with `valid_from`/`valid_to` timestamps and the originating `_operation`. + +Reconstruction: + +- `reconstruct_as_of(table, seq)` — returns the rows valid at a given sequence + (point-in-time / time travel) from SCD2 history. +- `restore_from_lake(target_seq)` — deterministically rebuilds the current snapshot by + replaying the durable lake up to a sequence (operational rollback). + +`apply_cdc_records` is **idempotent and out-of-order safe**: a record is skipped if the +row's current `_cdc_seq` is already ≥ its sequence. + +--- + +## 7. Schema-change safety (`ingestion/schema_contract.py`) + +Before every ingest cycle the live source schema is diffed against `SCHEMA_CONTRACT`: + +- **Breaking → fail closed:** dropped/renamed column, type change, NOT-NULL tightening, + or enum-domain shrink raises `SchemaIncompatibilityError` and **nothing is written**. +- **Additive → warn and continue:** brand-new columns or new enum values are reported but + do not stop ingestion. + +CI gate: `scripts/check_schema_contracts.py` (exits non-zero on a breaking change). + +--- + +## 8. Validation parity (`validation/checks.py`) + +The source's rules are re-asserted against the warehouse so downstream consumers are +protected even when the source is wrong: + +- **System checks:** PK uniqueness, NOT-NULL, referential integrity (FKs), enum domains. +- **Business checks:** `order.total = Σ line_total`, `line_total = quantity * + unit_price`, non-negative money, `price > 0`, `quantity > 0`, `settled_at ≥ + created_at`, and legal status transitions (validated from the ordered lake history). + +CI gate: `scripts/run_data_quality_checks.py`. + +--- + +## 9. Catalog exposure (`catalog/catalog.json`) + +Every lake and warehouse dataset is registered with layer, description, owner, +consumers, update cadence, query path, and per-column schema. + +CI gate: `scripts/validate_catalog.py`; a test also asserts the catalog stays in sync +with the live tables. + +--- + +## 10. Reliability matrix + +| Scenario | Handling | +|---|---| +| Duplicate / replayed event | Lake PK + `ON CONFLICT DO NOTHING`; warehouse skips `seq ≤ _cdc_seq` | +| Out-of-order event | Warehouse skips stale sequences; lake is keyed by sequence | +| Late-arriving update | Same `seq ≤ _cdc_seq` guard: a newer state already applied is never overwritten by an older one; the lake still records it | +| Retry after partial failure | Lake append + warehouse apply are idempotent; checkpoint advances only after both succeed, so a retry re-applies the same window safely | +| Delete | `delete` event → snapshot soft-delete + SCD2 version closed | +| Ingestion restart after checkpoint | Resume from persisted checkpoint; replay only `records_since(checkpoint)` | +| Incompatible schema change | Fail closed before any write | +| Recovery from historical data | `reconstruct_as_of` (SCD2) / `restore_from_lake` (replay from lake) | + +--- + +## 11. Testing strategy (Red → Blue → Green) + +`pytest -q` → **41 passing** tests across: + +- `test_source_constraints.py` — PK/NOT-NULL/enum/FK/positive-amount constraints +- `test_cdc.py` — insert/update/delete capture, before/after images, replay +- `test_lake.py` — full retention, idempotent append, no duplication on re-ingest +- `test_warehouse.py` — latest snapshot, soft delete, SCD2 time travel, restore, + out-of-order/duplicate safety +- `test_schema_safety.py` — breaking changes stop, additive changes warn +- `test_validation_parity.py` — system + business violations are caught +- `test_catalog.py` — catalog completeness and catalog ↔ live-table parity + +--- + +## 12. How to run + +Simulated (no external services): + +```bash +cd submission/ecommerce-cdc +python -m venv .venv && . .venv/bin/activate +pip install -r requirements.txt + +pytest -q # full suite (41 tests) +python -m ingestion.pipeline # end-to-end demo (ingest + time travel + restore) +python scripts/check_schema_contracts.py # schema-safety gate +python scripts/run_data_quality_checks.py # validation-parity gate +python scripts/validate_catalog.py # catalog gate +``` + +Real PostgreSQL source (optional, Docker — full walkthrough in +[TESTING_POSTGRES.md](./TESTING_POSTGRES.md)): + +```bash +docker-compose up -d # Postgres, schema auto-created +python scripts/pg_seed.py # insert realistic sample rows +python scripts/pg_sync.py # capture changes -> lake + warehouse + report +python scripts/pg_mutate.py # legal updates + a safe delete +python scripts/pg_sync.py # re-sync to see update/delete capture +``` + +--- + +## 13. Assumptions & trade-offs + +- **DuckDB stands in for object storage + a cloud warehouse** to keep the solution + runnable and dependency-free; the layering (lake vs warehouse) is what matters and maps + cleanly to Parquet/Delta + BigQuery/Snowflake. +- **Snapshot-diff CDC** (in the Postgres bridge) is a polling approximation of log-based + CDC; it is correct and idempotent, and the documented production path is Debezium / WAL + with the LSN replacing `sequence`. Nothing downstream changes. +- **Row images are JSON** in the lake for schema flexibility; a columnar layout would be + used at scale. +- **Checkpoint is per-consumer** to allow multiple independent downstream readers. diff --git a/submission/ecommerce-cdc/TESTING_POSTGRES.md b/submission/ecommerce-cdc/TESTING_POSTGRES.md new file mode 100644 index 0000000..27e2d9e --- /dev/null +++ b/submission/ecommerce-cdc/TESTING_POSTGRES.md @@ -0,0 +1,138 @@ +# Testing against a real PostgreSQL source (Docker) + +This walks you through running the pipeline against a **real Postgres** instead of the +in-process simulator: create tables, insert/update/delete rows, and watch the changes +flow into the DuckDB **lake** (full history) and **warehouse** (current snapshot + time +travel). + +The simulated path (`pytest`, `python -m ingestion.pipeline`) needs none of this — it +is fully covered without Postgres. This Postgres path is **optional** and not run in CI. + +## Prerequisites + +- Docker engine. On macOS without Docker Desktop you can use Colima: + ```bash + brew install colima docker docker-compose + colima start + ``` +- Python deps (the Postgres driver `psycopg` is included in `requirements.txt`): + ```bash + cd submission/ecommerce-cdc + python -m venv .venv && . .venv/bin/activate + pip install -r requirements.txt + ``` + +## 1. Start Postgres (auto-creates the schema) + +```bash +docker-compose up -d +``` + +`docker-compose.yml` boots `postgres:16-alpine` and runs `source/postgres_ddl.sql` on +first start, so the 5 source tables + enums + indexes are created automatically. + +## 2. See the tables + +```bash +docker exec kulu_ecom_pg psql -U kulu -d kulu_ecom -c "\dt" +docker exec kulu_ecom_pg psql -U kulu -d kulu_ecom -c "\d orders" +``` + +Open an interactive SQL shell any time with: + +```bash +docker exec -it kulu_ecom_pg psql -U kulu -d kulu_ecom +``` + +Connection details: host `localhost`, port `5432`, db `kulu_ecom`, user/pass `kulu`/`kulu`. + +## 3. Insert rows + +Either run the sample seeder: + +```bash +python scripts/pg_seed.py +``` + +…or insert manually: + +```bash +docker exec kulu_ecom_pg psql -U kulu -d kulu_ecom -c \ + "INSERT INTO customers (customer_id, name, email, status) + VALUES ('c9','Dana','dana@example.com','active');" +``` + +## 4. Sync changes into the lake + warehouse + +```bash +python scripts/pg_sync.py +``` + +This diffs Postgres against the previous sync (snapshot-diff CDC), writes every change +to the lake, refreshes the warehouse (current snapshot + SCD2 history), runs validation +parity, and prints a summary. Example after seeding: + +``` +=== sync === +captured: 8 insert, 0 update, 0 delete (8 applied) +=== lake (full history) === +8 events {'insert': 8} +=== warehouse current snapshot === + orders: [('o1', 'pending', Decimal('70.00'))] +=== validation parity === + all checks passed +``` + +## 5. Update / delete and re-sync (see CDC in action) + +Use the batch mutation script (legal status transitions, a price change, a safe delete): + +```bash +python scripts/pg_mutate.py +python scripts/pg_sync.py +``` + +…or do it by hand: + +```bash +docker exec kulu_ecom_pg psql -U kulu -d kulu_ecom -c \ + "UPDATE orders SET status='paid', updated_at=now() WHERE order_id='o1'; + DELETE FROM customers WHERE customer_id='c2';" + +python scripts/pg_sync.py +``` + +You'll see: +- the lake grows (now 10 events: 8 insert + 1 update + 1 delete) — **full history retained** +- `wh_orders` shows `o1` = `paid` — **latest snapshot** +- `wh_customers` live count drops by one — **soft delete** +- time travel prints the order's state at an earlier sequence vs now + +## 6. Inspect the lakehouse directly (optional) + +The lakehouse is a DuckDB file at `.data/lakehouse.duckdb`: + +```bash +python - <<'PY' +import duckdb +c = duckdb.connect(".data/lakehouse.duckdb") +print(c.execute("SELECT sequence, operation, table_name, primary_key FROM lake_cdc_events ORDER BY sequence").fetchall()) +print(c.execute("SELECT order_id, status, _deleted FROM wh_orders").fetchall()) +print(c.execute("SELECT order_id, status, _valid_from_seq, _valid_to_seq, _is_current FROM wh_orders_history ORDER BY _valid_from_seq").fetchall()) +PY +``` + +## How this maps to production + +The `scripts/pg_sync.py` polling diff is the documented **simulated CDC**. In production +it is replaced by Debezium / Postgres logical replication reading the WAL; the sequence +becomes the LSN. Everything downstream — lake, warehouse, SCD2, validation, catalog — +stays exactly the same. + +## Teardown + +```bash +docker-compose down -v # stop Postgres and remove its data volume +rm -rf .data # clear the local lakehouse file +colima stop # (optional) stop the Docker VM +``` diff --git a/submission/ecommerce-cdc/catalog/catalog.json b/submission/ecommerce-cdc/catalog/catalog.json new file mode 100644 index 0000000..29530be --- /dev/null +++ b/submission/ecommerce-cdc/catalog/catalog.json @@ -0,0 +1,161 @@ +{ + "catalog_version": "1.0", + "source_system": "ecommerce-oltp (PostgreSQL)", + "datasets": [ + { + "name": "lake_cdc_events", + "layer": "lake", + "description": "Append-only, idempotent log of every CDC event captured from the e-commerce source. Full change history for replay, audit, and point-in-time recovery.", + "owner": "data-platform", + "consumers": ["data-platform", "audit"], + "update_cadence": "real-time", + "query_path": "SELECT * FROM lake_cdc_events ORDER BY sequence", + "schema": { + "sequence": "INTEGER (PK) — monotonic capture offset / LSN analogue", + "operation": "VARCHAR — insert | update | delete", + "table_name": "VARCHAR — source table", + "primary_key": "VARCHAR — source row PK value", + "data": "VARCHAR (JSON) — row image at capture time", + "captured_at": "TIMESTAMP — UTC capture time" + } + }, + { + "name": "wh_customers", + "layer": "warehouse", + "description": "Current-state customer snapshot. Soft-deleted rows flagged with _deleted=true.", + "owner": "data-platform", + "consumers": ["analytics", "marketing"], + "update_cadence": "near-real-time", + "query_path": "SELECT * FROM wh_customers WHERE NOT _deleted", + "schema": { + "customer_id": "VARCHAR (PK)", + "name": "VARCHAR", + "email": "VARCHAR", + "status": "VARCHAR — active | suspended | closed", + "created_at": "TIMESTAMP", + "updated_at": "TIMESTAMP", + "_cdc_seq": "INTEGER — sequence of last applied event", + "_deleted": "BOOLEAN — soft-delete flag" + } + }, + { + "name": "wh_products", + "layer": "warehouse", + "description": "Current-state product catalog snapshot.", + "owner": "data-platform", + "consumers": ["analytics", "merchandising"], + "update_cadence": "near-real-time", + "query_path": "SELECT * FROM wh_products WHERE NOT _deleted", + "schema": { + "product_id": "VARCHAR (PK)", + "sku": "VARCHAR", + "name": "VARCHAR", + "price": "DECIMAL(18,2) — > 0", + "status": "VARCHAR — active | discontinued", + "created_at": "TIMESTAMP", + "updated_at": "TIMESTAMP", + "_cdc_seq": "INTEGER", + "_deleted": "BOOLEAN" + } + }, + { + "name": "wh_orders", + "layer": "warehouse", + "description": "Current-state order snapshot. total must equal the sum of its order_items.line_total.", + "owner": "data-platform", + "consumers": ["analytics", "finance"], + "update_cadence": "near-real-time", + "query_path": "SELECT * FROM wh_orders WHERE NOT _deleted", + "schema": { + "order_id": "VARCHAR (PK)", + "customer_id": "VARCHAR — FK to wh_customers", + "status": "VARCHAR — pending | paid | shipped | delivered | cancelled | refunded", + "total": "DECIMAL(18,2) — >= 0", + "currency": "VARCHAR", + "created_at": "TIMESTAMP", + "updated_at": "TIMESTAMP", + "_cdc_seq": "INTEGER", + "_deleted": "BOOLEAN" + } + }, + { + "name": "wh_order_items", + "layer": "warehouse", + "description": "Current-state order line items (weak entity bound to an order).", + "owner": "data-platform", + "consumers": ["analytics", "finance"], + "update_cadence": "near-real-time", + "query_path": "SELECT * FROM wh_order_items WHERE NOT _deleted", + "schema": { + "order_item_id": "VARCHAR (PK)", + "order_id": "VARCHAR — FK to wh_orders", + "product_id": "VARCHAR — FK to wh_products", + "quantity": "INTEGER — > 0", + "unit_price": "DECIMAL(18,2) — >= 0", + "line_total": "DECIMAL(18,2) — = quantity * unit_price", + "_cdc_seq": "INTEGER", + "_deleted": "BOOLEAN" + } + }, + { + "name": "wh_payments", + "layer": "warehouse", + "description": "Current-state payment snapshot (weak entity bound to an order).", + "owner": "data-platform", + "consumers": ["analytics", "finance"], + "update_cadence": "near-real-time", + "query_path": "SELECT * FROM wh_payments WHERE NOT _deleted", + "schema": { + "payment_id": "VARCHAR (PK)", + "order_id": "VARCHAR — FK to wh_orders", + "amount": "DECIMAL(18,2) — >= 0", + "method": "VARCHAR — card | wallet | bank_transfer", + "status": "VARCHAR — pending | settled | failed | refunded", + "created_at": "TIMESTAMP", + "settled_at": "TIMESTAMP — nullable; >= created_at", + "_cdc_seq": "INTEGER", + "_deleted": "BOOLEAN" + } + }, + { + "name": "wh_orders_history", + "layer": "warehouse", + "description": "SCD2 version history for orders. Enables point-in-time reconstruction via _valid_from_seq / _valid_to_seq.", + "owner": "data-platform", + "consumers": ["analytics", "audit"], + "update_cadence": "near-real-time", + "query_path": "SELECT * FROM wh_orders_history WHERE _valid_from_seq <= :seq AND (_valid_to_seq IS NULL OR _valid_to_seq > :seq)", + "schema": { + "order_id": "VARCHAR (PK part)", + "status": "VARCHAR", + "total": "DECIMAL(18,2)", + "_valid_from_seq": "INTEGER (PK part) — version opened at this sequence", + "_valid_to_seq": "INTEGER — version closed at this sequence (NULL if current)", + "_is_current": "BOOLEAN", + "_operation": "VARCHAR", + "valid_from": "TIMESTAMP", + "valid_to": "TIMESTAMP" + } + }, + { + "name": "wh_payments_history", + "layer": "warehouse", + "description": "SCD2 version history for payments for point-in-time reconstruction.", + "owner": "data-platform", + "consumers": ["analytics", "audit", "finance"], + "update_cadence": "near-real-time", + "query_path": "SELECT * FROM wh_payments_history WHERE _valid_from_seq <= :seq AND (_valid_to_seq IS NULL OR _valid_to_seq > :seq)", + "schema": { + "payment_id": "VARCHAR (PK part)", + "status": "VARCHAR", + "amount": "DECIMAL(18,2)", + "_valid_from_seq": "INTEGER (PK part)", + "_valid_to_seq": "INTEGER", + "_is_current": "BOOLEAN", + "_operation": "VARCHAR", + "valid_from": "TIMESTAMP", + "valid_to": "TIMESTAMP" + } + } + ] +} diff --git a/submission/ecommerce-cdc/conftest.py b/submission/ecommerce-cdc/conftest.py new file mode 100644 index 0000000..cd7892b --- /dev/null +++ b/submission/ecommerce-cdc/conftest.py @@ -0,0 +1,6 @@ +"""Root conftest — puts the candidate dir on sys.path so packages import cleanly.""" + +import os +import sys + +sys.path.insert(0, os.path.dirname(__file__)) diff --git a/submission/ecommerce-cdc/docker-compose.yml b/submission/ecommerce-cdc/docker-compose.yml new file mode 100644 index 0000000..2aeccef --- /dev/null +++ b/submission/ecommerce-cdc/docker-compose.yml @@ -0,0 +1,18 @@ +services: + postgres: + image: postgres:16-alpine + container_name: kulu_ecom_pg + environment: + POSTGRES_USER: kulu + POSTGRES_PASSWORD: kulu + POSTGRES_DB: kulu_ecom + ports: + - "5432:5432" + volumes: + # Auto-creates the source schema on first boot. + - ./source/postgres_ddl.sql:/docker-entrypoint-initdb.d/01_schema.sql:ro + healthcheck: + test: ["CMD-SHELL", "pg_isready -U kulu -d kulu_ecom"] + interval: 3s + timeout: 3s + retries: 20 diff --git a/submission/ecommerce-cdc/ingestion/__init__.py b/submission/ecommerce-cdc/ingestion/__init__.py new file mode 100644 index 0000000..2dfae31 --- /dev/null +++ b/submission/ecommerce-cdc/ingestion/__init__.py @@ -0,0 +1 @@ +"""Ingestion layer: CDC capture, checkpointing, schema safety, orchestration.""" diff --git a/submission/ecommerce-cdc/ingestion/cdc.py b/submission/ecommerce-cdc/ingestion/cdc.py new file mode 100644 index 0000000..f22f47a --- /dev/null +++ b/submission/ecommerce-cdc/ingestion/cdc.py @@ -0,0 +1,117 @@ +""" +CDC capture layer. + +Simulates WAL/logical-replication change capture: every insert/update/delete on the +source produces a ``CDCRecord`` with a monotonically increasing ``sequence`` (the +local analogue of a Kafka offset or Postgres LSN) plus before/after row images. + +Replay safety: a consumer checkpoints the last processed ``sequence`` and calls +``records_since(offset)`` to replay only unprocessed changes after a restart. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Any, Dict, List, Optional + +VALID_OPERATIONS = frozenset({"insert", "update", "delete"}) + + +def _utcnow() -> datetime: + return datetime.now(timezone.utc) + + +@dataclass +class CDCRecord: + operation: str + table: str + primary_key: str + before: Optional[Dict[str, Any]] + after: Optional[Dict[str, Any]] + captured_at: datetime = field(default_factory=_utcnow) + sequence: int = 0 + + def __post_init__(self) -> None: + if self.operation not in VALID_OPERATIONS: + raise ValueError( + f"Invalid CDC operation {self.operation!r}. " + f"Must be one of: {sorted(VALID_OPERATIONS)}" + ) + + @property + def image(self) -> Dict[str, Any]: + """Row image to persist: ``after`` for insert/update, ``before`` for delete.""" + return (self.after if self.after is not None else self.before) or {} + + +class CDCCapture: + """ + In-process CDC log. + + Production analogue: a Debezium/Kafka connector reading the Postgres WAL. Each + record carries a ``sequence`` equivalent to a Kafka offset / Postgres LSN used for + checkpoint-based replay. + """ + + def __init__(self, start_sequence: int = 0) -> None: + self._log: List[CDCRecord] = [] + self._seq: int = start_sequence + + # -- write API ------------------------------------------------------------ + + def insert(self, table: str, pk: str, after: Dict[str, Any]) -> CDCRecord: + return self._record("insert", table, pk, None, after) + + def update( + self, + table: str, + pk: str, + after: Dict[str, Any], + before: Optional[Dict[str, Any]] = None, + ) -> CDCRecord: + return self._record("update", table, pk, before, after) + + def delete( + self, + table: str, + pk: str, + before: Optional[Dict[str, Any]] = None, + ) -> CDCRecord: + return self._record("delete", table, pk, before, None) + + # -- read / replay API ---------------------------------------------------- + + def records_since(self, offset: int = 0) -> List[CDCRecord]: + """Return all records with ``sequence`` > offset (checkpoint replay).""" + return [r for r in self._log if r.sequence > offset] + + @property + def latest_sequence(self) -> int: + return self._seq + + @property + def log(self) -> List[CDCRecord]: + return list(self._log) + + # -- internal ------------------------------------------------------------- + + def _record( + self, + operation: str, + table: str, + pk: str, + before: Optional[Dict[str, Any]], + after: Optional[Dict[str, Any]], + ) -> CDCRecord: + self._seq += 1 + rec = CDCRecord( + operation=operation, + table=table, + primary_key=pk, + before=before, + after=after, + sequence=self._seq, + ) + self._log.append(rec) + return rec diff --git a/submission/ecommerce-cdc/ingestion/checkpoint.py b/submission/ecommerce-cdc/ingestion/checkpoint.py new file mode 100644 index 0000000..76febab --- /dev/null +++ b/submission/ecommerce-cdc/ingestion/checkpoint.py @@ -0,0 +1,50 @@ +""" +Checkpoint store. + +Persists the last successfully applied CDC ``sequence`` so ingestion can restart +without losing or reprocessing changes. Backed by a DuckDB table so the checkpoint +shares the warehouse's durability guarantees. + +Production analogue: a Kafka consumer group offset or a Postgres replication-slot +confirmed_flush_lsn. +""" + +from __future__ import annotations + +import duckdb + +_CHECKPOINT_TABLE = "_ingest_checkpoint" + + +def create_checkpoint_table(conn: duckdb.DuckDBPyConnection) -> None: + conn.execute( + f""" + CREATE TABLE IF NOT EXISTS {_CHECKPOINT_TABLE} ( + consumer VARCHAR PRIMARY KEY, + sequence INTEGER NOT NULL + ) + """ + ) + + +def get_checkpoint(conn: duckdb.DuckDBPyConnection, consumer: str = "default") -> int: + create_checkpoint_table(conn) + row = conn.execute( + f"SELECT sequence FROM {_CHECKPOINT_TABLE} WHERE consumer = ?", + [consumer], + ).fetchone() + return int(row[0]) if row else 0 + + +def set_checkpoint( + conn: duckdb.DuckDBPyConnection, sequence: int, consumer: str = "default" +) -> None: + create_checkpoint_table(conn) + conn.execute( + f""" + INSERT INTO {_CHECKPOINT_TABLE} (consumer, sequence) + VALUES (?, ?) + ON CONFLICT (consumer) DO UPDATE SET sequence = excluded.sequence + """, + [consumer, sequence], + ) diff --git a/submission/ecommerce-cdc/ingestion/pipeline.py b/submission/ecommerce-cdc/ingestion/pipeline.py new file mode 100644 index 0000000..6d3e0f3 --- /dev/null +++ b/submission/ecommerce-cdc/ingestion/pipeline.py @@ -0,0 +1,105 @@ +""" +Ingestion orchestrator. + +Ties the layers together for one ingest cycle: + + 1. schema-contract guard (fail closed on incompatible source change) + 2. read unprocessed records since the checkpoint + 3. append every change to the lake (idempotent) + 4. apply changes to the warehouse (current snapshot + SCD2 history) + 5. advance the checkpoint only after a successful cycle + +Running it twice is a no-op (idempotent); restarting after a crash resumes from the +last checkpoint. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Dict, List, Optional, Tuple + +import duckdb + +from ingestion.cdc import CDCCapture +from ingestion.checkpoint import create_checkpoint_table, get_checkpoint, set_checkpoint +from ingestion.schema_contract import assert_compatible +from lake.lake import append_to_lake, create_lake_table +from source.models import create_source_tables +from warehouse.warehouse import apply_cdc_records, create_warehouse_tables + + +@dataclass +class IngestionResult: + applied: int + lake_written: int + checkpoint: int + warnings: Dict[str, List[str]] = field(default_factory=dict) + + +def build_lakehouse(conn: duckdb.DuckDBPyConnection) -> None: + """Create source, lake, warehouse, and checkpoint objects on one connection.""" + create_source_tables(conn) + create_lake_table(conn) + create_warehouse_tables(conn) + create_checkpoint_table(conn) + + +def run_ingestion( + conn: duckdb.DuckDBPyConnection, + capture: CDCCapture, + consumer: str = "default", + enforce_schema: bool = True, + observed_enums: Optional[Dict[str, Dict[str, Tuple[str, ...]]]] = None, +) -> IngestionResult: + warnings: Dict[str, List[str]] = {} + if enforce_schema: + # Raises SchemaIncompatibilityError on a breaking change -> nothing is written. + warnings = assert_compatible(conn, observed_enums=observed_enums) + + checkpoint = get_checkpoint(conn, consumer) + records = capture.records_since(checkpoint) + if not records: + return IngestionResult(0, 0, checkpoint, warnings) + + lake_written = append_to_lake(conn, records) + apply_cdc_records(conn, records) + + new_checkpoint = capture.latest_sequence + set_checkpoint(conn, new_checkpoint, consumer) + return IngestionResult(len(records), lake_written, new_checkpoint, warnings) + + +def _demo() -> None: + from ingestion.seed import seed_capture + from warehouse.warehouse import reconstruct_as_of, restore_from_lake + + conn = duckdb.connect(":memory:") + build_lakehouse(conn) + capture = seed_capture() + + result = run_ingestion(conn, capture) + print( + f"Ingested {result.applied} change(s); " + f"{result.lake_written} written to lake; checkpoint={result.checkpoint}" + ) + + lake_total = conn.execute("SELECT COUNT(*) FROM lake_cdc_events").fetchone()[0] + print(f"Lake holds {lake_total} immutable change events.") + + current = conn.execute( + "SELECT order_id, status, total FROM wh_orders WHERE NOT _deleted" + ).fetchall() + print(f"Current orders snapshot: {current}") + + early = reconstruct_as_of(conn, "orders", as_of_seq=9) + print(f"Order o1 status at seq 9 (time travel): {early[0]['status']}") + + restore_from_lake(conn, target_seq=9) + restored = conn.execute( + "SELECT status FROM wh_orders WHERE order_id = 'o1'" + ).fetchone() + print(f"After restore_from_lake(seq=9), order o1 status: {restored[0]}") + + +if __name__ == "__main__": + _demo() diff --git a/submission/ecommerce-cdc/ingestion/schema_contract.py b/submission/ecommerce-cdc/ingestion/schema_contract.py new file mode 100644 index 0000000..ade8aa3 --- /dev/null +++ b/submission/ecommerce-cdc/ingestion/schema_contract.py @@ -0,0 +1,130 @@ +""" +Schema-change detection and safe stop. + +Source systems give no backward-compatibility guarantee, so before every ingest cycle +we diff the live source schema against the declared ``SCHEMA_CONTRACT``. Incompatible +changes stop ingestion (fail closed); additive changes only warn. + +Incompatible (breaking): +- dropped column (also catches renames, which look like drop + add) +- changed data type of a contracted column +- a contracted NOT NULL column that became nullable in the source +- shrunk enum domain (an allowed value was removed) + +Compatible (additive, warn only): +- brand-new columns not in the contract +- new enum values added to an existing domain +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Dict, List, Optional, Tuple + +import duckdb + +from source.models import SCHEMA_CONTRACT, TableContract + + +class SchemaIncompatibilityError(RuntimeError): + """Raised to stop the line when an incompatible source change is detected.""" + + +@dataclass +class SchemaDiff: + breaking: List[str] + additive: List[str] + + +def observed_schema( + conn: duckdb.DuckDBPyConnection, table: str +) -> Dict[str, Tuple[str, bool]]: + """Return {column_name: (dtype, nullable)} as reported by the live source.""" + rows = conn.execute(f"DESCRIBE {table}").fetchall() + schema: Dict[str, Tuple[str, bool]] = {} + for row in rows: + col_name = row[0] + col_type = str(row[1]).upper() + nullable = str(row[2]).upper() == "YES" + schema[col_name] = (col_type, nullable) + return schema + + +def compare_table( + contract: TableContract, + observed: Dict[str, Tuple[str, bool]], + observed_enums: Optional[Dict[str, Tuple[str, ...]]] = None, +) -> SchemaDiff: + """Diff one table's contract against an observed schema.""" + breaking: List[str] = [] + additive: List[str] = [] + observed_enums = observed_enums or {} + + for spec in contract.columns: + if spec.name not in observed: + breaking.append(f"column '{spec.name}' dropped or renamed") + continue + + obs_type, obs_nullable = observed[spec.name] + if obs_type != spec.dtype.upper(): + breaking.append( + f"column '{spec.name}' type changed: " f"{spec.dtype} -> {obs_type}" + ) + if not spec.nullable and obs_nullable: + breaking.append( + f"column '{spec.name}' became nullable but contract requires NOT NULL" + ) + + if spec.enum is not None and spec.name in observed_enums: + removed = set(spec.enum) - set(observed_enums[spec.name]) + added = set(observed_enums[spec.name]) - set(spec.enum) + if removed: + breaking.append( + f"column '{spec.name}' enum value(s) removed: {sorted(removed)}" + ) + if added: + additive.append( + f"column '{spec.name}' enum value(s) added: {sorted(added)}" + ) + + for col in observed: + if contract.column(col) is None: + additive.append(f"new column '{col}' not in contract") + + return SchemaDiff(breaking=breaking, additive=additive) + + +def assert_compatible( + conn: duckdb.DuckDBPyConnection, + contract: Dict[str, TableContract] = SCHEMA_CONTRACT, + observed_enums: Optional[Dict[str, Dict[str, Tuple[str, ...]]]] = None, +) -> Dict[str, List[str]]: + """ + Check every contracted table against the live source schema. + + Returns a map of table -> additive-change warnings. + Raises ``SchemaIncompatibilityError`` if any breaking change is detected. + """ + observed_enums = observed_enums or {} + all_breaking: List[str] = [] + warnings: Dict[str, List[str]] = {} + + for table, table_contract in contract.items(): + try: + observed = observed_schema(conn, table) + except Exception as exc: # noqa: BLE001 - surfaced as a breaking change + all_breaking.append(f"{table}: cannot read source schema ({exc})") + continue + + diff = compare_table(table_contract, observed, observed_enums.get(table)) + all_breaking.extend(f"{table}: {msg}" for msg in diff.breaking) + if diff.additive: + warnings[table] = diff.additive + + if all_breaking: + raise SchemaIncompatibilityError( + "Incompatible source schema change(s) detected; ingestion stopped:\n - " + + "\n - ".join(all_breaking) + ) + + return warnings diff --git a/submission/ecommerce-cdc/ingestion/seed.py b/submission/ecommerce-cdc/ingestion/seed.py new file mode 100644 index 0000000..bc31ae0 --- /dev/null +++ b/submission/ecommerce-cdc/ingestion/seed.py @@ -0,0 +1,121 @@ +""" +Representative seed scenario shared by the demo, tests, and quality scripts. + +The scenario exercises every CDC path: inserts across all five tables, an update +(order pending -> paid, payment pending -> settled), and a delete (customer c3). +Sequences are deterministic so time-travel assertions are stable: + + seq 1- 9 inserts (c1,c2,c3, p1,p2, o1, oi1,oi2, pay1) + seq 10 update order o1 -> paid + seq 11 update payment pay1 -> settled + seq 12 delete customer c3 +""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from typing import Optional + +from ingestion.cdc import CDCCapture + +BASE_TS = datetime(2026, 1, 1, 12, 0, 0, tzinfo=timezone.utc) +LATER_TS = BASE_TS + timedelta(hours=1) + + +def seed_capture() -> CDCCapture: + cap = CDCCapture() + + cap.insert("customers", "c1", _customer("c1", "Alice", "alice@example.com")) + cap.insert("customers", "c2", _customer("c2", "Bob", "bob@example.com")) + cap.insert("customers", "c3", _customer("c3", "Carol", "carol@example.com")) + + cap.insert("products", "p1", _product("p1", "SKU-1", "Keyboard", 30.00)) + cap.insert("products", "p2", _product("p2", "SKU-2", "Mouse", 10.00)) + + cap.insert("orders", "o1", _order("o1", "c1", "pending", 70.00)) + cap.insert("order_items", "oi1", _item("oi1", "o1", "p1", 2, 30.00, 60.00)) + cap.insert("order_items", "oi2", _item("oi2", "o1", "p2", 1, 10.00, 10.00)) + cap.insert("payments", "pay1", _payment("pay1", "o1", 70.00, "card", "pending")) + + cap.update( + "orders", + "o1", + _order("o1", "c1", "paid", 70.00, updated_at=LATER_TS), + before=_order("o1", "c1", "pending", 70.00), + ) + cap.update( + "payments", + "pay1", + _payment("pay1", "o1", 70.00, "card", "settled", settled_at=LATER_TS), + before=_payment("pay1", "o1", 70.00, "card", "pending"), + ) + cap.delete("customers", "c3", before=_customer("c3", "Carol", "carol@example.com")) + + return cap + + +def _customer(cid: str, name: str, email: str) -> dict: + return { + "customer_id": cid, + "name": name, + "email": email, + "status": "active", + "created_at": BASE_TS, + "updated_at": BASE_TS, + } + + +def _product(pid: str, sku: str, name: str, price: float) -> dict: + return { + "product_id": pid, + "sku": sku, + "name": name, + "price": price, + "status": "active", + "created_at": BASE_TS, + "updated_at": BASE_TS, + } + + +def _order( + oid: str, cid: str, status: str, total: float, updated_at: datetime = BASE_TS +) -> dict: + return { + "order_id": oid, + "customer_id": cid, + "status": status, + "total": total, + "currency": "USD", + "created_at": BASE_TS, + "updated_at": updated_at, + } + + +def _item(iid: str, oid: str, pid: str, qty: int, unit: float, line: float) -> dict: + return { + "order_item_id": iid, + "order_id": oid, + "product_id": pid, + "quantity": qty, + "unit_price": unit, + "line_total": line, + } + + +def _payment( + pid: str, + oid: str, + amount: float, + method: str, + status: str, + settled_at: Optional[datetime] = None, +) -> dict: + return { + "payment_id": pid, + "order_id": oid, + "amount": amount, + "method": method, + "status": status, + "created_at": BASE_TS, + "settled_at": settled_at, + } diff --git a/submission/ecommerce-cdc/lake/__init__.py b/submission/ecommerce-cdc/lake/__init__.py new file mode 100644 index 0000000..de44ee1 --- /dev/null +++ b/submission/ecommerce-cdc/lake/__init__.py @@ -0,0 +1 @@ +"""Lake layer: append-only, idempotent CDC change history.""" diff --git a/submission/ecommerce-cdc/lake/lake.py b/submission/ecommerce-cdc/lake/lake.py new file mode 100644 index 0000000..866b34e --- /dev/null +++ b/submission/ecommerce-cdc/lake/lake.py @@ -0,0 +1,114 @@ +""" +Lake layer — append-only storage for every CDC event. + +Every change is written exactly once. The lake is the source of truth for replay and +historical reconstruction; rows are never updated or deleted. + +Idempotency: ``sequence`` is the primary key and appends use ``ON CONFLICT DO NOTHING``, +so duplicate or replayed events cannot corrupt the history. + +Production analogue: Parquet/Delta files on object storage, partitioned by +``table_name`` and capture date. +""" + +from __future__ import annotations + +import json +from datetime import datetime +from decimal import Decimal +from typing import Any, Dict, List, Optional + +import duckdb + +from ingestion.cdc import CDCRecord + + +def create_lake_table(conn: duckdb.DuckDBPyConnection) -> None: + conn.execute( + """ + CREATE TABLE IF NOT EXISTS lake_cdc_events ( + sequence INTEGER PRIMARY KEY, + operation VARCHAR NOT NULL, + table_name VARCHAR NOT NULL, + primary_key VARCHAR NOT NULL, + data VARCHAR NOT NULL, + captured_at TIMESTAMP NOT NULL + ) + """ + ) + + +def append_to_lake(conn: duckdb.DuckDBPyConnection, records: List[CDCRecord]) -> int: + """ + Append CDC records to the lake idempotently. + + Returns the number of newly written records (duplicates by ``sequence`` are + silently skipped). + """ + if not records: + return 0 + + before = _row_count(conn) + for r in records: + conn.execute( + """ + INSERT INTO lake_cdc_events + (sequence, operation, table_name, primary_key, data, captured_at) + VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT (sequence) DO NOTHING + """, + [ + r.sequence, + r.operation, + r.table, + r.primary_key, + _serialize(r.image), + r.captured_at, + ], + ) + return _row_count(conn) - before + + +def read_events( + conn: duckdb.DuckDBPyConnection, up_to_sequence: Optional[int] = None +) -> List[Dict[str, Any]]: + """Read lake events in sequence order, optionally up to ``up_to_sequence``.""" + if up_to_sequence is None: + rows = conn.execute( + "SELECT sequence, operation, table_name, primary_key, data, captured_at " + "FROM lake_cdc_events ORDER BY sequence" + ).fetchall() + else: + rows = conn.execute( + "SELECT sequence, operation, table_name, primary_key, data, captured_at " + "FROM lake_cdc_events WHERE sequence <= ? ORDER BY sequence", + [up_to_sequence], + ).fetchall() + + return [ + { + "sequence": row[0], + "operation": row[1], + "table_name": row[2], + "primary_key": row[3], + "data": json.loads(row[4]), + "captured_at": row[5], + } + for row in rows + ] + + +def _row_count(conn: duckdb.DuckDBPyConnection) -> int: + return conn.execute("SELECT COUNT(*) FROM lake_cdc_events").fetchone()[0] + + +def _serialize(data: Dict[str, Any]) -> str: + return json.dumps(data, default=_json_default) + + +def _json_default(obj: object) -> str: + if isinstance(obj, datetime): + return obj.isoformat() + if isinstance(obj, Decimal): + return str(obj) + raise TypeError(f"Object of type {type(obj).__name__} is not JSON serializable") diff --git a/submission/ecommerce-cdc/postgres/__init__.py b/submission/ecommerce-cdc/postgres/__init__.py new file mode 100644 index 0000000..2aeedfb --- /dev/null +++ b/submission/ecommerce-cdc/postgres/__init__.py @@ -0,0 +1,6 @@ +"""Optional PostgreSQL source integration (requires psycopg + a running Postgres). + +This package lets you exercise the pipeline against a real Postgres source instead of +the in-process simulator: insert/update/delete rows in Postgres, then run a sync that +captures the changes (snapshot-diff CDC) into the DuckDB lake + warehouse. +""" diff --git a/submission/ecommerce-cdc/postgres/bridge.py b/submission/ecommerce-cdc/postgres/bridge.py new file mode 100644 index 0000000..0fda83b --- /dev/null +++ b/submission/ecommerce-cdc/postgres/bridge.py @@ -0,0 +1,169 @@ +""" +PostgreSQL -> lakehouse CDC bridge (snapshot-diff capture). + +On each ``sync`` we read the current state of every source table from Postgres and +compare it to the last-seen snapshot stored in DuckDB. The differences become CDC +records (insert / update / delete) that are fed through the existing lake + warehouse +machinery, with a sequence that continues monotonically across syncs. + +This is the documented "simulated CDC" approach made real against Postgres. In +production this polling diff is replaced by Debezium / logical replication reading the +WAL; everything downstream (lake, warehouse, SCD2, validation) stays identical. +""" + +from __future__ import annotations + +import json +import os +from datetime import datetime +from decimal import Decimal +from typing import Any, Dict, Tuple + +import duckdb + +from ingestion.cdc import CDCCapture +from ingestion.checkpoint import ( + create_checkpoint_table, + get_checkpoint, + set_checkpoint, +) +from lake.lake import append_to_lake, create_lake_table +from postgres.config import LAKEHOUSE_PATH, pg_dsn +from source.models import SCHEMA_CONTRACT +from warehouse.warehouse import apply_cdc_records, create_warehouse_tables + +SYNC_ORDER = ["customers", "products", "orders", "order_items", "payments"] +_SNAPSHOT_TABLE = "_pg_snapshot" + + +def open_lakehouse() -> duckdb.DuckDBPyConnection: + """Open (and initialize) the persistent DuckDB lakehouse.""" + os.makedirs(os.path.dirname(LAKEHOUSE_PATH), exist_ok=True) + conn = duckdb.connect(LAKEHOUSE_PATH) + create_lake_table(conn) + create_warehouse_tables(conn) + create_checkpoint_table(conn) + _create_snapshot_table(conn) + return conn + + +def connect_postgres(): + import psycopg # imported lazily so the core stays dependency-free + + return psycopg.connect(pg_dsn()) + + +def sync(pg_conn, duck_conn: duckdb.DuckDBPyConnection) -> Dict[str, int]: + """Capture Postgres changes since the last sync into the lake + warehouse.""" + # Restart marker shared with the simulated pipeline: resume sequencing from the + # last committed checkpoint. In steady state this equals MAX(lake sequence); we + # guard with that max so a crash between lake-write and checkpoint can't reuse + # a sequence (lake append is also idempotent on sequence). + lake_max = duck_conn.execute( + "SELECT COALESCE(MAX(sequence), 0) FROM lake_cdc_events" + ).fetchone()[0] + start_seq = max(int(get_checkpoint(duck_conn)), int(lake_max)) + cap = CDCCapture(start_sequence=start_seq) + counts = {"insert": 0, "update": 0, "delete": 0} + + for table in SYNC_ORDER: + contract = SCHEMA_CONTRACT[table] + pk = contract.primary_key + current = _read_pg_table(pg_conn, table, contract.column_names, pk) + previous = _read_snapshot(duck_conn, table) + + for key, (row, row_json) in current.items(): + if key not in previous: + cap.insert(table, key, row) + counts["insert"] += 1 + elif previous[key] != row_json: + cap.update(table, key, row, before=json.loads(previous[key])) + counts["update"] += 1 + + for key, old_json in previous.items(): + if key not in current: + cap.delete(table, key, before=json.loads(old_json)) + counts["delete"] += 1 + + records = cap.records_since(start_seq) + if records: + append_to_lake(duck_conn, records) + apply_cdc_records(duck_conn, records) + set_checkpoint(duck_conn, cap.latest_sequence) + _store_snapshot(duck_conn, current_by_table=_snapshot_state(pg_conn)) + + counts["applied"] = len(records) + return counts + + +# -- internal helpers --------------------------------------------------------- + + +def _read_pg_table( + pg_conn, table: str, columns: list, pk: str +) -> Dict[str, Tuple[Dict[str, Any], str]]: + col_list = ", ".join(columns) + with pg_conn.cursor() as cur: + cur.execute(f"SELECT {col_list} FROM {table}") + rows = cur.fetchall() + result: Dict[str, Tuple[Dict[str, Any], str]] = {} + for row in rows: + record = dict(zip(columns, row)) + result[str(record[pk])] = (record, _row_json(record)) + return result + + +def _snapshot_state(pg_conn) -> Dict[str, Dict[str, str]]: + state: Dict[str, Dict[str, str]] = {} + for table in SYNC_ORDER: + contract = SCHEMA_CONTRACT[table] + rows = _read_pg_table( + pg_conn, table, contract.column_names, contract.primary_key + ) + state[table] = {pk: row_json for pk, (_, row_json) in rows.items()} + return state + + +def _create_snapshot_table(conn: duckdb.DuckDBPyConnection) -> None: + conn.execute( + f""" + CREATE TABLE IF NOT EXISTS {_SNAPSHOT_TABLE} ( + table_name VARCHAR NOT NULL, + pk VARCHAR NOT NULL, + row_json VARCHAR NOT NULL, + PRIMARY KEY (table_name, pk) + ) + """ + ) + + +def _read_snapshot(conn: duckdb.DuckDBPyConnection, table: str) -> Dict[str, str]: + rows = conn.execute( + f"SELECT pk, row_json FROM {_SNAPSHOT_TABLE} WHERE table_name = ?", + [table], + ).fetchall() + return {row[0]: row[1] for row in rows} + + +def _store_snapshot( + conn: duckdb.DuckDBPyConnection, current_by_table: Dict[str, Dict[str, str]] +) -> None: + conn.execute(f"DELETE FROM {_SNAPSHOT_TABLE}") + for table, rows in current_by_table.items(): + for pk, row_json in rows.items(): + conn.execute( + f"INSERT INTO {_SNAPSHOT_TABLE} VALUES (?, ?, ?)", + [table, pk, row_json], + ) + + +def _row_json(record: Dict[str, Any]) -> str: + return json.dumps(record, sort_keys=True, default=_json_default) + + +def _json_default(obj: object) -> str: + if isinstance(obj, datetime): + return obj.isoformat() + if isinstance(obj, Decimal): + return str(obj) + return str(obj) diff --git a/submission/ecommerce-cdc/postgres/config.py b/submission/ecommerce-cdc/postgres/config.py new file mode 100644 index 0000000..9fa60c9 --- /dev/null +++ b/submission/ecommerce-cdc/postgres/config.py @@ -0,0 +1,30 @@ +"""Connection/config for the optional Postgres source integration. + +Defaults match docker-compose.yml; override with standard PG* env vars. +""" + +from __future__ import annotations + +import os + +PG_CONN = { + "host": os.environ.get("PGHOST", "localhost"), + "port": int(os.environ.get("PGPORT", "5432")), + "user": os.environ.get("PGUSER", "kulu"), + "password": os.environ.get("PGPASSWORD", "kulu"), + "dbname": os.environ.get("PGDATABASE", "kulu_ecom"), +} + +# Persistent DuckDB lakehouse file so lake/warehouse state survives across syncs. +_SUBMISSION_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +LAKEHOUSE_PATH = os.environ.get( + "LAKEHOUSE_DB", os.path.join(_SUBMISSION_ROOT, ".data", "lakehouse.duckdb") +) + + +def pg_dsn() -> str: + c = PG_CONN + return ( + f"host={c['host']} port={c['port']} user={c['user']} " + f"password={c['password']} dbname={c['dbname']}" + ) diff --git a/submission/ecommerce-cdc/requirements.txt b/submission/ecommerce-cdc/requirements.txt new file mode 100644 index 0000000..9c4fb55 --- /dev/null +++ b/submission/ecommerce-cdc/requirements.txt @@ -0,0 +1,7 @@ +# Core — pipeline + full test suite run with just these (no external services). +duckdb>=0.10.0 +pytest>=7.4 + +# Optional — only for the real PostgreSQL source path (scripts/pg_*.py). +# Not needed for the simulated pipeline or tests. See TESTING_POSTGRES.md. +psycopg[binary]>=3.1 diff --git a/submission/ecommerce-cdc/scripts/check_schema_contracts.py b/submission/ecommerce-cdc/scripts/check_schema_contracts.py new file mode 100644 index 0000000..5e08e18 --- /dev/null +++ b/submission/ecommerce-cdc/scripts/check_schema_contracts.py @@ -0,0 +1,44 @@ +#!/usr/bin/env python3 +""" +check_schema_contracts.py + +CI gate for schema safety. Builds the source schema and asserts it matches the +declared SCHEMA_CONTRACT. A breaking change (dropped/renamed column, type change, +nullable tightening, shrunk enum) stops the build before ingestion can drift. + +Exit 0 — contract satisfied (additive-only changes are reported as warnings). +Exit 1 — incompatible change detected. +""" + +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +import duckdb + +from ingestion.schema_contract import SchemaIncompatibilityError, assert_compatible +from source.models import SCHEMA_CONTRACT, create_source_tables + + +def main() -> int: + conn = duckdb.connect(":memory:") + create_source_tables(conn) + + try: + warnings = assert_compatible(conn) + except SchemaIncompatibilityError as exc: + print("Schema contract check FAILED:") + print(exc) + return 1 + + for table, msgs in warnings.items(): + for msg in msgs: + print(f" ! {table}: {msg} (additive — allowed)") + + print(f"All schema contracts passed ({len(SCHEMA_CONTRACT)} tables checked).") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/submission/ecommerce-cdc/scripts/pg_mutate.py b/submission/ecommerce-cdc/scripts/pg_mutate.py new file mode 100644 index 0000000..162e066 --- /dev/null +++ b/submission/ecommerce-cdc/scripts/pg_mutate.py @@ -0,0 +1,75 @@ +#!/usr/bin/env python3 +""" +pg_mutate.py — apply a batch of realistic changes to the Postgres source. + +Demonstrates CDC change-capture without typing SQL: updates (legal status +transitions, a price change, a customer reactivation) and a safe delete. Run this, +then run scripts/pg_sync.py to see the updates/deletes flow into the lake + warehouse. + +All changes keep the warehouse valid: +- status transitions are legal (pending->paid, paid->shipped, shipped->delivered, ...) +- the deleted customer (c6) has no orders, so no referential orphans are created +- a product price change does not affect historical order totals (orders keep their + captured unit_price), so order.total == sum(line_total) still holds + +Re-running is safe but not a no-op: each UPDATE bumps updated_at, so the row genuinely +changes and is re-captured as an update on the next sync. Status stays the same, so the +status-transition check treats it as a no-change and validation still passes. +""" + +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from postgres.bridge import connect_postgres + +# (description, SQL) +MUTATIONS = [ + ( + "order o3: pending -> paid", + "UPDATE orders SET status='paid', updated_at=now() WHERE order_id='o3'", + ), + ( + "payment pay3: pending -> settled", + "UPDATE payments SET status='settled', settled_at=now() " + "WHERE payment_id='pay3'", + ), + ( + "order o1: pending -> paid", + "UPDATE orders SET status='paid', updated_at=now() WHERE order_id='o1'", + ), + ( + "order o2: shipped -> delivered", + "UPDATE orders SET status='delivered', updated_at=now() WHERE order_id='o2'", + ), + ( + "customer c4: suspended -> active (reactivated)", + "UPDATE customers SET status='active', updated_at=now() WHERE customer_id='c4'", + ), + ( + "product p2: price 10.00 -> 11.50", + "UPDATE products SET price=11.50, updated_at=now() WHERE product_id='p2'", + ), + ( + "customer c6: deleted (no orders -> safe)", + "DELETE FROM customers WHERE customer_id='c6'", + ), +] + + +def main() -> int: + conn = connect_postgres() + with conn: + with conn.cursor() as cur: + for desc, sql in MUTATIONS: + cur.execute(sql) + print(f" applied: {desc} (rows affected: {cur.rowcount})") + conn.close() + print(f"\nApplied {len(MUTATIONS)} change(s) to Postgres.") + print("Next: python scripts/pg_sync.py") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/submission/ecommerce-cdc/scripts/pg_seed.py b/submission/ecommerce-cdc/scripts/pg_seed.py new file mode 100644 index 0000000..e3970bc --- /dev/null +++ b/submission/ecommerce-cdc/scripts/pg_seed.py @@ -0,0 +1,133 @@ +#!/usr/bin/env python3 +""" +pg_seed.py — insert a realistic set of sample rows into the Postgres source. + +Idempotent (ON CONFLICT DO NOTHING) so you can run it repeatedly. After seeding, +run scripts/pg_sync.py to capture the changes into the lake + warehouse. + +Invariants are kept intact so validation parity passes: + line_total == quantity * unit_price, and order.total == sum(line_total). +""" + +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from postgres.bridge import connect_postgres + +# (customer_id, name, email, status) +CUSTOMERS = [ + ("c1", "Alice Chen", "alice@example.com", "active"), + ("c2", "Bob Martinez", "bob@example.com", "active"), + ("c3", "Carol Nguyen", "carol@example.com", "active"), + ("c4", "Dan OBrien", "dan@example.com", "suspended"), + ("c5", "Eve Kowalski", "eve@example.com", "active"), + ("c6", "Frank Muller", "frank@example.com", "closed"), +] + +# (product_id, sku, name, price, status) +PRODUCTS = [ + ("p1", "SKU-KEYB", "Mechanical Keyboard", 30.00, "active"), + ("p2", "SKU-MOUSE", "Wireless Mouse", 10.00, "active"), + ("p3", "SKU-MON27", "27in 4K Monitor", 199.99, "active"), + ("p4", "SKU-USBC", "USB-C Cable 2m", 12.50, "active"), + ("p5", "SKU-CAM", "1080p Webcam", 59.00, "active"), + ("p6", "SKU-MAT", "Desk Mat XL", 25.00, "discontinued"), +] + +# (order_id, customer_id, status, currency, [(item_id, product_id, qty, unit_price)]) +ORDERS = [ + ("o1", "c1", "pending", "USD", [("oi1", "p1", 2, 30.00), ("oi2", "p2", 1, 10.00)]), + ("o2", "c3", "shipped", "USD", [("oi3", "p3", 1, 199.99), ("oi4", "p4", 2, 12.50)]), + ("o3", "c4", "pending", "EUR", [("oi5", "p5", 1, 59.00)]), + ( + "o4", + "c5", + "delivered", + "USD", + [("oi6", "p4", 3, 12.50), ("oi7", "p1", 1, 30.00)], + ), + ("o5", "c1", "cancelled", "USD", [("oi8", "p5", 2, 59.00)]), + ("o6", "c2", "paid", "GBP", [("oi9", "p2", 4, 10.00), ("oi10", "p6", 1, 25.00)]), +] + +# (payment_id, order_id, amount, method, status, settled) +PAYMENTS = [ + ("pay1", "o1", 70.00, "card", "pending", False), + ("pay2", "o2", 224.99, "card", "settled", True), + ("pay3", "o3", 59.00, "wallet", "pending", False), + ("pay4", "o4", 67.50, "bank_transfer", "settled", True), + ("pay5", "o5", 118.00, "card", "failed", False), + ("pay6", "o6", 65.00, "wallet", "settled", True), +] + + +def build_rows() -> list: + rows = [] + + for cid, name, email, status in CUSTOMERS: + rows.append( + "INSERT INTO customers (customer_id, name, email, status) " + f"VALUES ('{cid}','{name}','{email}','{status}') ON CONFLICT DO NOTHING" + ) + + for pid, sku, name, price, status in PRODUCTS: + rows.append( + "INSERT INTO products (product_id, sku, name, price, status) " + f"VALUES ('{pid}','{sku}','{name}',{price},'{status}') " + "ON CONFLICT DO NOTHING" + ) + + for oid, cid, status, currency, items in ORDERS: + total = round(sum(qty * unit for _, _, qty, unit in items), 2) + rows.append( + "INSERT INTO orders (order_id, customer_id, status, total, currency) " + f"VALUES ('{oid}','{cid}','{status}',{total},'{currency}') " + "ON CONFLICT DO NOTHING" + ) + for item_id, pid, qty, unit in items: + line = round(qty * unit, 2) + rows.append( + "INSERT INTO order_items (order_item_id, order_id, product_id, " + "quantity, unit_price, line_total) " + f"VALUES ('{item_id}','{oid}','{pid}',{qty},{unit},{line}) " + "ON CONFLICT DO NOTHING" + ) + + for pid, oid, amount, method, status, settled in PAYMENTS: + if settled: + rows.append( + "INSERT INTO payments (payment_id, order_id, amount, method, status, " + "created_at, settled_at) " + f"VALUES ('{pid}','{oid}',{amount},'{method}','{status}',now(),now()) " + "ON CONFLICT DO NOTHING" + ) + else: + rows.append( + "INSERT INTO payments (payment_id, order_id, amount, method, status) " + f"VALUES ('{pid}','{oid}',{amount},'{method}','{status}') " + "ON CONFLICT DO NOTHING" + ) + + return rows + + +def main() -> int: + rows = build_rows() + conn = connect_postgres() + with conn: + with conn.cursor() as cur: + for sql in rows: + cur.execute(sql) + conn.close() + print( + f"Seeded {len(CUSTOMERS)} customers, {len(PRODUCTS)} products, " + f"{len(ORDERS)} orders, {len(PAYMENTS)} payments into Postgres." + ) + print("Next: python scripts/pg_sync.py") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/submission/ecommerce-cdc/scripts/pg_sync.py b/submission/ecommerce-cdc/scripts/pg_sync.py new file mode 100644 index 0000000..ff8f58f --- /dev/null +++ b/submission/ecommerce-cdc/scripts/pg_sync.py @@ -0,0 +1,72 @@ +#!/usr/bin/env python3 +""" +pg_sync.py — capture Postgres changes into the lake + warehouse, then report state. + +Run this after inserting/updating/deleting rows in Postgres. It diffs the source +against the last sync, writes the changes to the lake, refreshes the warehouse +(current snapshot + SCD2 history), runs validation parity, and prints a summary. +""" + +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from postgres.bridge import connect_postgres, open_lakehouse, sync +from validation.checks import run_all_checks +from warehouse.warehouse import reconstruct_as_of + + +def main() -> int: + duck = open_lakehouse() + pg = connect_postgres() + try: + counts = sync(pg, duck) + finally: + pg.close() + + print("=== sync ===") + print( + f"captured: {counts['insert']} insert, {counts['update']} update, " + f"{counts['delete']} delete ({counts['applied']} applied)" + ) + + lake_total = duck.execute("SELECT COUNT(*) FROM lake_cdc_events").fetchone()[0] + by_op = dict( + duck.execute( + "SELECT operation, COUNT(*) FROM lake_cdc_events GROUP BY operation" + ).fetchall() + ) + print(f"\n=== lake (full history) ===\n{lake_total} events {by_op}") + + print("\n=== warehouse current snapshot ===") + for table in ("customers", "products", "orders", "order_items", "payments"): + rows = duck.execute( + f"SELECT COUNT(*) FROM wh_{table} WHERE NOT _deleted" + ).fetchone()[0] + print(f" wh_{table}: {rows} live row(s)") + + orders = duck.execute( + "SELECT order_id, status, total FROM wh_orders WHERE NOT _deleted " + "ORDER BY order_id" + ).fetchall() + print(f" orders: {orders}") + + max_seq = duck.execute( + "SELECT COALESCE(MAX(sequence), 0) FROM lake_cdc_events" + ).fetchone()[0] + if max_seq: + first = reconstruct_as_of(duck, "orders", as_of_seq=1) + print(f"\n=== time travel ===\norders as of seq 1: {first}") + latest = reconstruct_as_of(duck, "orders", as_of_seq=max_seq) + print(f"orders as of seq {max_seq}: {latest}") + + failures = run_all_checks(duck) + print("\n=== validation parity ===") + print(" all checks passed" if not failures else f" FAILURES: {failures}") + duck.close() + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/submission/ecommerce-cdc/scripts/run_data_quality_checks.py b/submission/ecommerce-cdc/scripts/run_data_quality_checks.py new file mode 100644 index 0000000..a277886 --- /dev/null +++ b/submission/ecommerce-cdc/scripts/run_data_quality_checks.py @@ -0,0 +1,42 @@ +#!/usr/bin/env python3 +""" +run_data_quality_checks.py + +Seeds the lakehouse with the representative scenario, runs the full ingest cycle, then +asserts system + business validation parity against the warehouse outputs. + +Exit 0 — all checks pass. +Exit 1 — one or more failures found. +""" + +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +import duckdb + +from ingestion.pipeline import build_lakehouse, run_ingestion +from ingestion.seed import seed_capture +from validation.checks import run_all_checks + + +def main() -> int: + conn = duckdb.connect(":memory:") + build_lakehouse(conn) + run_ingestion(conn, seed_capture()) + + failures = run_all_checks(conn) + + if failures: + print("Data quality failures:") + for f in failures: + print(f" x {f}") + return 1 + + print("All data quality checks passed (system + business parity).") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/submission/ecommerce-cdc/scripts/validate_catalog.py b/submission/ecommerce-cdc/scripts/validate_catalog.py new file mode 100644 index 0000000..08784c2 --- /dev/null +++ b/submission/ecommerce-cdc/scripts/validate_catalog.py @@ -0,0 +1,86 @@ +#!/usr/bin/env python3 +""" +validate_catalog.py + +Verifies catalog/catalog.json exists and has a complete entry for every required lake +and warehouse dataset, with all required metadata fields populated. + +Exit 0 — catalog is valid. +Exit 1 — missing file, missing datasets, or missing required fields. +""" + +import json +import os +import sys + +CATALOG_PATH = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "catalog", + "catalog.json", +) + +REQUIRED_DATASETS = [ + "lake_cdc_events", + "wh_customers", + "wh_products", + "wh_orders", + "wh_order_items", + "wh_payments", + "wh_orders_history", + "wh_payments_history", +] + +REQUIRED_FIELDS = [ + "name", + "layer", + "description", + "owner", + "schema", + "update_cadence", + "query_path", +] + + +def validate() -> list: + violations = [] + + if not os.path.exists(CATALOG_PATH): + violations.append(f"catalog.json not found at {CATALOG_PATH}") + return violations + + with open(CATALOG_PATH) as f: + try: + catalog = json.load(f) + except json.JSONDecodeError as exc: + violations.append(f"catalog.json is not valid JSON: {exc}") + return violations + + datasets = {d["name"]: d for d in catalog.get("datasets", []) if "name" in d} + + for name in REQUIRED_DATASETS: + if name not in datasets: + violations.append(f"Missing dataset entry: {name}") + continue + entry = datasets[name] + for field in REQUIRED_FIELDS: + if field not in entry or not entry[field]: + violations.append(f"{name}: missing or empty required field '{field}'") + + return violations + + +def main() -> int: + violations = validate() + + if violations: + print("Catalog validation failures:") + for v in violations: + print(f" x {v}") + return 1 + + print(f"Catalog validation passed ({len(REQUIRED_DATASETS)} datasets verified).") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/submission/ecommerce-cdc/source/__init__.py b/submission/ecommerce-cdc/source/__init__.py new file mode 100644 index 0000000..d8bfb8f --- /dev/null +++ b/submission/ecommerce-cdc/source/__init__.py @@ -0,0 +1 @@ +"""Source layer: canonical e-commerce schema, contract, and seed data.""" diff --git a/submission/ecommerce-cdc/source/models.py b/submission/ecommerce-cdc/source/models.py new file mode 100644 index 0000000..0a1aab8 --- /dev/null +++ b/submission/ecommerce-cdc/source/models.py @@ -0,0 +1,264 @@ +""" +Source schema for an e-commerce order system. + +Domain: customers place orders; orders contain order_items (referencing products) +and are paid via payments. + +Strong entities : customers, products, orders +Weak entities : order_items (identity completed by its parent order), + payments (lifecycle bound to its parent order) + +The runnable pipeline materializes these tables in DuckDB so the whole solution is +dependency-free. ``source/postgres_ddl.sql`` holds the canonical PostgreSQL design +(constraints + indexes) that this mirrors. + +Invariants: +- order.total == sum(order_items.line_total) for the order +- line_total == quantity * unit_price +- monetary amounts >= 0; price > 0; quantity > 0 +- settled_at >= created_at when present +- enum/status fields restricted to known domains +- valid status transitions only +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Dict, List, Optional, Tuple + +import duckdb + + +@dataclass(frozen=True) +class ColumnSpec: + """A single column's contract: logical type, nullability, optional enum domain.""" + + name: str + dtype: str + nullable: bool = False + enum: Optional[Tuple[str, ...]] = None + + +@dataclass(frozen=True) +class TableContract: + """Declared shape of a source table used for schema-drift detection.""" + + primary_key: str + columns: List[ColumnSpec] + foreign_keys: Dict[str, Tuple[str, str]] = field(default_factory=dict) + + @property + def column_names(self) -> List[str]: + return [c.name for c in self.columns] + + def column(self, name: str) -> Optional[ColumnSpec]: + for c in self.columns: + if c.name == name: + return c + return None + + @property + def enum_domains(self) -> Dict[str, Tuple[str, ...]]: + return {c.name: c.enum for c in self.columns if c.enum is not None} + + +# Allowed business status transitions (used by validation parity). +ORDER_STATUS_TRANSITIONS: Dict[str, Tuple[str, ...]] = { + "pending": ("paid", "cancelled"), + "paid": ("shipped", "cancelled", "refunded"), + "shipped": ("delivered",), + "delivered": (), + "cancelled": (), + "refunded": (), +} + +PAYMENT_STATUS_TRANSITIONS: Dict[str, Tuple[str, ...]] = { + "pending": ("settled", "failed"), + "settled": ("refunded",), + "failed": (), + "refunded": (), +} + + +SCHEMA_CONTRACT: Dict[str, TableContract] = { + "customers": TableContract( + primary_key="customer_id", + columns=[ + ColumnSpec("customer_id", "VARCHAR"), + ColumnSpec("name", "VARCHAR"), + ColumnSpec("email", "VARCHAR"), + ColumnSpec( + "status", + "VARCHAR", + enum=("active", "suspended", "closed"), + ), + ColumnSpec("created_at", "TIMESTAMP"), + ColumnSpec("updated_at", "TIMESTAMP"), + ], + ), + "products": TableContract( + primary_key="product_id", + columns=[ + ColumnSpec("product_id", "VARCHAR"), + ColumnSpec("sku", "VARCHAR"), + ColumnSpec("name", "VARCHAR"), + ColumnSpec("price", "DECIMAL(18,2)"), + ColumnSpec("status", "VARCHAR", enum=("active", "discontinued")), + ColumnSpec("created_at", "TIMESTAMP"), + ColumnSpec("updated_at", "TIMESTAMP"), + ], + ), + "orders": TableContract( + primary_key="order_id", + foreign_keys={"customer_id": ("customers", "customer_id")}, + columns=[ + ColumnSpec("order_id", "VARCHAR"), + ColumnSpec("customer_id", "VARCHAR"), + ColumnSpec( + "status", + "VARCHAR", + enum=( + "pending", + "paid", + "shipped", + "delivered", + "cancelled", + "refunded", + ), + ), + ColumnSpec("total", "DECIMAL(18,2)"), + ColumnSpec("currency", "VARCHAR"), + ColumnSpec("created_at", "TIMESTAMP"), + ColumnSpec("updated_at", "TIMESTAMP"), + ], + ), + "order_items": TableContract( + primary_key="order_item_id", + foreign_keys={ + "order_id": ("orders", "order_id"), + "product_id": ("products", "product_id"), + }, + columns=[ + ColumnSpec("order_item_id", "VARCHAR"), + ColumnSpec("order_id", "VARCHAR"), + ColumnSpec("product_id", "VARCHAR"), + ColumnSpec("quantity", "INTEGER"), + ColumnSpec("unit_price", "DECIMAL(18,2)"), + ColumnSpec("line_total", "DECIMAL(18,2)"), + ], + ), + "payments": TableContract( + primary_key="payment_id", + foreign_keys={"order_id": ("orders", "order_id")}, + columns=[ + ColumnSpec("payment_id", "VARCHAR"), + ColumnSpec("order_id", "VARCHAR"), + ColumnSpec("amount", "DECIMAL(18,2)"), + ColumnSpec( + "method", + "VARCHAR", + enum=("card", "wallet", "bank_transfer"), + ), + ColumnSpec( + "status", + "VARCHAR", + enum=("pending", "settled", "failed", "refunded"), + ), + ColumnSpec("created_at", "TIMESTAMP"), + ColumnSpec("settled_at", "TIMESTAMP", nullable=True), + ], + ), +} + + +def create_source_tables(conn: duckdb.DuckDBPyConnection) -> None: + """Create the source tables (with constraints) in the given DuckDB connection.""" + conn.execute( + """ + CREATE TABLE IF NOT EXISTS customers ( + customer_id VARCHAR PRIMARY KEY, + name VARCHAR NOT NULL, + email VARCHAR NOT NULL, + status VARCHAR NOT NULL + CHECK (status IN ('active', 'suspended', 'closed')), + created_at TIMESTAMP NOT NULL, + updated_at TIMESTAMP NOT NULL + ) + """ + ) + + conn.execute( + """ + CREATE TABLE IF NOT EXISTS products ( + product_id VARCHAR PRIMARY KEY, + sku VARCHAR NOT NULL, + name VARCHAR NOT NULL, + price DECIMAL(18, 2) NOT NULL CHECK (price > 0), + status VARCHAR NOT NULL + CHECK (status IN ('active', 'discontinued')), + created_at TIMESTAMP NOT NULL, + updated_at TIMESTAMP NOT NULL + ) + """ + ) + + conn.execute( + """ + CREATE TABLE IF NOT EXISTS orders ( + order_id VARCHAR PRIMARY KEY, + customer_id VARCHAR NOT NULL REFERENCES customers(customer_id), + status VARCHAR NOT NULL + CHECK (status IN ('pending', 'paid', 'shipped', + 'delivered', 'cancelled', 'refunded')), + total DECIMAL(18, 2) NOT NULL CHECK (total >= 0), + currency VARCHAR NOT NULL, + created_at TIMESTAMP NOT NULL, + updated_at TIMESTAMP NOT NULL + ) + """ + ) + + conn.execute( + """ + CREATE TABLE IF NOT EXISTS order_items ( + order_item_id VARCHAR PRIMARY KEY, + order_id VARCHAR NOT NULL REFERENCES orders(order_id), + product_id VARCHAR NOT NULL REFERENCES products(product_id), + quantity INTEGER NOT NULL CHECK (quantity > 0), + unit_price DECIMAL(18, 2) NOT NULL CHECK (unit_price >= 0), + line_total DECIMAL(18, 2) NOT NULL CHECK (line_total >= 0) + ) + """ + ) + + conn.execute( + """ + CREATE TABLE IF NOT EXISTS payments ( + payment_id VARCHAR PRIMARY KEY, + order_id VARCHAR NOT NULL REFERENCES orders(order_id), + amount DECIMAL(18, 2) NOT NULL CHECK (amount >= 0), + method VARCHAR NOT NULL + CHECK (method IN ('card', 'wallet', 'bank_transfer')), + status VARCHAR NOT NULL + CHECK (status IN ('pending', 'settled', 'failed', 'refunded')), + created_at TIMESTAMP NOT NULL, + settled_at TIMESTAMP + ) + """ + ) + + _create_indexes(conn) + + +def _create_indexes(conn: duckdb.DuckDBPyConnection) -> None: + """Indexes on FK and change-capture (updated_at) columns.""" + statements = [ + "CREATE INDEX IF NOT EXISTS idx_orders_customer ON orders(customer_id)", + "CREATE INDEX IF NOT EXISTS idx_orders_updated ON orders(updated_at)", + "CREATE INDEX IF NOT EXISTS idx_items_order ON order_items(order_id)", + "CREATE INDEX IF NOT EXISTS idx_items_product ON order_items(product_id)", + "CREATE INDEX IF NOT EXISTS idx_payments_order ON payments(order_id)", + "CREATE INDEX IF NOT EXISTS idx_customers_updated ON customers(updated_at)", + ] + for stmt in statements: + conn.execute(stmt) diff --git a/submission/ecommerce-cdc/source/postgres_ddl.sql b/submission/ecommerce-cdc/source/postgres_ddl.sql new file mode 100644 index 0000000..4f73ae2 --- /dev/null +++ b/submission/ecommerce-cdc/source/postgres_ddl.sql @@ -0,0 +1,72 @@ +-- Canonical PostgreSQL source schema for the e-commerce CDC pipeline. +-- The runnable pipeline mirrors this in DuckDB (see source/models.py); in +-- production this is the system of record and CDC is read from its WAL. + +CREATE TYPE customer_status AS ENUM ('active', 'suspended', 'closed'); +CREATE TYPE product_status AS ENUM ('active', 'discontinued'); +CREATE TYPE order_status AS ENUM ('pending', 'paid', 'shipped', + 'delivered', 'cancelled', 'refunded'); +CREATE TYPE payment_method AS ENUM ('card', 'wallet', 'bank_transfer'); +CREATE TYPE payment_status AS ENUM ('pending', 'settled', 'failed', 'refunded'); + +-- Strong entity: customers +CREATE TABLE customers ( + customer_id TEXT PRIMARY KEY, + name TEXT NOT NULL, + email TEXT NOT NULL, + status customer_status NOT NULL DEFAULT 'active', + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +-- Strong entity: products +CREATE TABLE products ( + product_id TEXT PRIMARY KEY, + sku TEXT NOT NULL UNIQUE, + name TEXT NOT NULL, + price NUMERIC(18, 2) NOT NULL CHECK (price > 0), + status product_status NOT NULL DEFAULT 'active', + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +-- Strong entity: orders +CREATE TABLE orders ( + order_id TEXT PRIMARY KEY, + customer_id TEXT NOT NULL REFERENCES customers(customer_id), + status order_status NOT NULL DEFAULT 'pending', + total NUMERIC(18, 2) NOT NULL CHECK (total >= 0), + currency TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +-- Weak entity: order_items (identity completed by parent order) +CREATE TABLE order_items ( + order_item_id TEXT PRIMARY KEY, + order_id TEXT NOT NULL REFERENCES orders(order_id), + product_id TEXT NOT NULL REFERENCES products(product_id), + quantity INTEGER NOT NULL CHECK (quantity > 0), + unit_price NUMERIC(18, 2) NOT NULL CHECK (unit_price >= 0), + line_total NUMERIC(18, 2) NOT NULL CHECK (line_total >= 0) +); + +-- Weak entity: payments (lifecycle bound to parent order) +CREATE TABLE payments ( + payment_id TEXT PRIMARY KEY, + order_id TEXT NOT NULL REFERENCES orders(order_id), + amount NUMERIC(18, 2) NOT NULL CHECK (amount >= 0), + method payment_method NOT NULL, + status payment_status NOT NULL DEFAULT 'pending', + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + settled_at TIMESTAMPTZ, + CHECK (settled_at IS NULL OR settled_at >= created_at) +); + +-- Indexes on FK and change-capture (updated_at) columns. +CREATE INDEX idx_orders_customer ON orders(customer_id); +CREATE INDEX idx_orders_updated ON orders(updated_at); +CREATE INDEX idx_items_order ON order_items(order_id); +CREATE INDEX idx_items_product ON order_items(product_id); +CREATE INDEX idx_payments_order ON payments(order_id); +CREATE INDEX idx_customers_updated ON customers(updated_at); diff --git a/submission/ecommerce-cdc/tests/conftest.py b/submission/ecommerce-cdc/tests/conftest.py new file mode 100644 index 0000000..0937a38 --- /dev/null +++ b/submission/ecommerce-cdc/tests/conftest.py @@ -0,0 +1,29 @@ +"""Shared fixtures for the e-commerce CDC lakehouse tests.""" + +import duckdb +import pytest + +from ingestion.cdc import CDCCapture +from ingestion.pipeline import build_lakehouse, run_ingestion +from ingestion.seed import seed_capture + + +@pytest.fixture() +def conn() -> duckdb.DuckDBPyConnection: + """Fresh in-memory lakehouse: source + lake + warehouse + checkpoint.""" + c = duckdb.connect(":memory:") + build_lakehouse(c) + return c + + +@pytest.fixture() +def capture() -> CDCCapture: + """The deterministic representative scenario (not yet ingested).""" + return seed_capture() + + +@pytest.fixture() +def ingested(conn, capture): + """Lakehouse after one full ingest cycle of the seed scenario.""" + run_ingestion(conn, capture) + return conn diff --git a/submission/ecommerce-cdc/tests/test_catalog.py b/submission/ecommerce-cdc/tests/test_catalog.py new file mode 100644 index 0000000..2e67b8f --- /dev/null +++ b/submission/ecommerce-cdc/tests/test_catalog.py @@ -0,0 +1,44 @@ +"""Catalog tests: entries exist for every lake & warehouse dataset with metadata.""" + +import json +import os + +from scripts.validate_catalog import REQUIRED_DATASETS, REQUIRED_FIELDS, validate + +CATALOG_PATH = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "catalog", + "catalog.json", +) + + +def _catalog(): + with open(CATALOG_PATH) as f: + return json.load(f) + + +def test_catalog_validates(): + assert validate() == [] + + +def test_every_required_dataset_present(): + names = {d["name"] for d in _catalog()["datasets"]} + for required in REQUIRED_DATASETS: + assert required in names + + +def test_each_entry_has_required_fields(): + for entry in _catalog()["datasets"]: + for field in REQUIRED_FIELDS: + assert entry.get(field), f"{entry.get('name')} missing {field}" + + +def test_layers_are_lake_or_warehouse(): + for entry in _catalog()["datasets"]: + assert entry["layer"] in {"lake", "warehouse"} + + +def test_catalog_matches_live_warehouse_tables(ingested): + live = {row[0] for row in ingested.execute("SHOW TABLES").fetchall()} + for entry in _catalog()["datasets"]: + assert entry["name"] in live, f"{entry['name']} not found in lakehouse" diff --git a/submission/ecommerce-cdc/tests/test_cdc.py b/submission/ecommerce-cdc/tests/test_cdc.py new file mode 100644 index 0000000..8cf46c5 --- /dev/null +++ b/submission/ecommerce-cdc/tests/test_cdc.py @@ -0,0 +1,55 @@ +"""CDC correctness tests: capture, replay, restart, duplicates.""" + +import pytest + +from ingestion.cdc import CDCCapture, CDCRecord + + +def test_invalid_operation_rejected(): + with pytest.raises(ValueError): + CDCRecord("upsert", "customers", "c1", None, {}) + + +def test_insert_update_delete_captured(): + cap = CDCCapture() + cap.insert("customers", "c1", {"customer_id": "c1"}) + cap.update("customers", "c1", {"customer_id": "c1", "name": "new"}) + cap.delete("customers", "c1", before={"customer_id": "c1"}) + + ops = [r.operation for r in cap.log] + assert ops == ["insert", "update", "delete"] + assert [r.sequence for r in cap.log] == [1, 2, 3] + + +def test_before_after_images(): + cap = CDCCapture() + rec = cap.update("orders", "o1", {"status": "paid"}, before={"status": "pending"}) + assert rec.before == {"status": "pending"} + assert rec.after == {"status": "paid"} + assert rec.image == {"status": "paid"} + + dele = cap.delete("orders", "o1", before={"status": "paid"}) + assert dele.after is None + assert dele.image == {"status": "paid"} # delete image falls back to before + + +def test_records_since_checkpoint_replay(): + cap = CDCCapture() + for i in range(5): + cap.insert("customers", f"c{i}", {"customer_id": f"c{i}"}) + + tail = cap.records_since(3) + assert [r.sequence for r in tail] == [4, 5] + assert cap.records_since(cap.latest_sequence) == [] + + +def test_restart_resumes_from_checkpoint(): + cap = CDCCapture() + cap.insert("customers", "c1", {"customer_id": "c1"}) + cap.insert("customers", "c2", {"customer_id": "c2"}) + + checkpoint = cap.latest_sequence + cap.insert("customers", "c3", {"customer_id": "c3"}) + + new = cap.records_since(checkpoint) + assert [r.primary_key for r in new] == ["c3"] diff --git a/submission/ecommerce-cdc/tests/test_lake.py b/submission/ecommerce-cdc/tests/test_lake.py new file mode 100644 index 0000000..851b290 --- /dev/null +++ b/submission/ecommerce-cdc/tests/test_lake.py @@ -0,0 +1,51 @@ +"""Lake tests: full change retention + idempotent append.""" + +from ingestion.cdc import CDCCapture +from ingestion.pipeline import run_ingestion +from lake.lake import append_to_lake, read_events + + +def test_lake_retains_every_change(ingested): + # 12 seed events: 9 inserts + 2 updates + 1 delete. + total = ingested.execute("SELECT COUNT(*) FROM lake_cdc_events").fetchone()[0] + assert total == 12 + + ops = ingested.execute( + "SELECT operation, COUNT(*) FROM lake_cdc_events GROUP BY operation" + ).fetchall() + counts = dict(ops) + assert counts == {"insert": 9, "update": 2, "delete": 1} + + +def test_lake_append_is_idempotent(conn): + cap = CDCCapture() + cap.insert("customers", "c1", {"customer_id": "c1"}) + records = cap.records_since(0) + + assert append_to_lake(conn, records) == 1 + assert append_to_lake(conn, records) == 0 # duplicate replay is a no-op + + total = conn.execute("SELECT COUNT(*) FROM lake_cdc_events").fetchone()[0] + assert total == 1 + + +def test_rerunning_ingestion_does_not_duplicate(conn): + run_ingestion(conn, _seed()) + # A second run with the same capture re-reads from the advanced checkpoint -> no-op. + cap = _seed() + result = run_ingestion(conn, cap) + assert result.applied == 0 + total = conn.execute("SELECT COUNT(*) FROM lake_cdc_events").fetchone()[0] + assert total == 12 + + +def test_read_events_up_to_sequence(ingested): + events = read_events(ingested, up_to_sequence=6) + assert all(e["sequence"] <= 6 for e in events) + assert len(events) == 6 + + +def _seed(): + from ingestion.seed import seed_capture + + return seed_capture() diff --git a/submission/ecommerce-cdc/tests/test_schema_safety.py b/submission/ecommerce-cdc/tests/test_schema_safety.py new file mode 100644 index 0000000..497aa75 --- /dev/null +++ b/submission/ecommerce-cdc/tests/test_schema_safety.py @@ -0,0 +1,74 @@ +"""Schema-change safety tests: detection, fail-closed stop, additive tolerance.""" + +import duckdb +import pytest + +from ingestion.cdc import CDCCapture +from ingestion.pipeline import build_lakehouse, run_ingestion +from ingestion.schema_contract import ( + SchemaIncompatibilityError, + assert_compatible, + compare_table, + observed_schema, +) +from source.models import SCHEMA_CONTRACT + + +def test_clean_schema_is_compatible(conn): + warnings = assert_compatible(conn) + assert warnings == {} + + +def test_dropped_column_is_breaking(conn): + # payments has no dependents; drop its index before mutating columns. + conn.execute("DROP INDEX IF EXISTS idx_payments_order") + conn.execute("ALTER TABLE payments DROP COLUMN method") + with pytest.raises(SchemaIncompatibilityError) as exc: + assert_compatible(conn) + assert "method" in str(exc.value) + + +def test_type_change_is_breaking(conn): + conn.execute("DROP INDEX IF EXISTS idx_payments_order") + conn.execute("ALTER TABLE payments ALTER COLUMN created_at TYPE VARCHAR") + with pytest.raises(SchemaIncompatibilityError) as exc: + assert_compatible(conn) + assert "created_at" in str(exc.value) + + +def test_additive_column_is_allowed(conn): + conn.execute("ALTER TABLE customers ADD COLUMN loyalty_tier VARCHAR") + warnings = assert_compatible(conn) + assert "customers" in warnings + assert any("loyalty_tier" in w for w in warnings["customers"]) + + +def test_enum_shrink_is_breaking(): + contract = SCHEMA_CONTRACT["orders"] + observed = {c.name: (c.dtype.upper(), c.nullable) for c in contract.columns} + shrunk = {"status": ("pending", "paid")} # dropped several states + diff = compare_table(contract, observed, observed_enums=shrunk) + assert any("enum value(s) removed" in m for m in diff.breaking) + + +def test_ingestion_stops_and_writes_nothing_on_breaking_change(conn): + conn.execute("DROP INDEX IF EXISTS idx_payments_order") + conn.execute("ALTER TABLE payments DROP COLUMN method") + + cap = CDCCapture() + cap.insert("customers", "c1", {"customer_id": "c1"}) + + with pytest.raises(SchemaIncompatibilityError): + run_ingestion(conn, cap) + + # Fail closed: nothing written to lake or warehouse, checkpoint not advanced. + lake = conn.execute("SELECT COUNT(*) FROM lake_cdc_events").fetchone()[0] + assert lake == 0 + + +def test_observed_schema_reports_nullability(): + c = duckdb.connect(":memory:") + build_lakehouse(c) + schema = observed_schema(c, "payments") + assert schema["settled_at"][1] is True # nullable + assert schema["payment_id"][1] is False # NOT NULL diff --git a/submission/ecommerce-cdc/tests/test_source_constraints.py b/submission/ecommerce-cdc/tests/test_source_constraints.py new file mode 100644 index 0000000..b52f577 --- /dev/null +++ b/submission/ecommerce-cdc/tests/test_source_constraints.py @@ -0,0 +1,63 @@ +"""Modeling & constraint tests for the source layer.""" + +import duckdb +import pytest + +from source.models import SCHEMA_CONTRACT, create_source_tables + + +@pytest.fixture() +def source() -> duckdb.DuckDBPyConnection: + c = duckdb.connect(":memory:") + create_source_tables(c) + return c + + +def _insert_customer(conn, cid="c1", status="active"): + conn.execute( + "INSERT INTO customers VALUES (?, 'A', 'a@x.com', ?, now(), now())", + [cid, status], + ) + + +def test_primary_key_uniqueness(source): + _insert_customer(source, "c1") + with pytest.raises(duckdb.ConstraintException): + _insert_customer(source, "c1") + + +def test_not_null_enforced(source): + with pytest.raises(duckdb.ConstraintException): + source.execute( + "INSERT INTO customers VALUES ('c2', NULL, 'a@x.com', " + "'active', now(), now())" + ) + + +def test_enum_check_constraint(source): + with pytest.raises(duckdb.ConstraintException): + _insert_customer(source, "c3", status="banana") + + +def test_foreign_key_reference(source): + with pytest.raises(duckdb.ConstraintException): + source.execute( + "INSERT INTO orders VALUES " + "('o1', 'ghost', 'pending', 10.00, 'USD', now(), now())" + ) + + +def test_positive_amount_constraints(source): + _insert_customer(source, "c1") + with pytest.raises(duckdb.ConstraintException): + source.execute( + "INSERT INTO products VALUES " + "('p1', 'sku', 'x', -1.00, 'active', now(), now())" + ) + + +def test_contract_matches_created_columns(source): + for table, contract in SCHEMA_CONTRACT.items(): + described = {row[0] for row in source.execute(f"DESCRIBE {table}").fetchall()} + for col in contract.column_names: + assert col in described, f"{table}.{col} missing from source DDL" diff --git a/submission/ecommerce-cdc/tests/test_validation_parity.py b/submission/ecommerce-cdc/tests/test_validation_parity.py new file mode 100644 index 0000000..640b15d --- /dev/null +++ b/submission/ecommerce-cdc/tests/test_validation_parity.py @@ -0,0 +1,75 @@ +"""Validation parity tests: system + business rules mirrored in the warehouse.""" + +from validation.checks import run_all_checks, run_business_checks, run_system_checks + + +def test_clean_warehouse_passes_all_checks(ingested): + assert run_all_checks(ingested) == [] + + +def test_negative_payment_amount_is_caught(ingested): + ingested.execute("UPDATE wh_payments SET amount = -5 WHERE payment_id = 'pay1'") + failures = run_business_checks(ingested) + assert any("negative amount" in f for f in failures) + + +def test_order_total_mismatch_is_caught(ingested): + ingested.execute("UPDATE wh_orders SET total = 999 WHERE order_id = 'o1'") + failures = run_business_checks(ingested) + assert any("sum(order_items.line_total)" in f for f in failures) + + +def test_line_total_mismatch_is_caught(ingested): + ingested.execute( + "UPDATE wh_order_items SET line_total = 1 WHERE order_item_id = 'oi1'" + ) + failures = run_business_checks(ingested) + assert any("line_total" in f for f in failures) + + +def test_referential_integrity_violation_is_caught(ingested): + ingested.execute("UPDATE wh_orders SET customer_id = 'ghost' WHERE order_id = 'o1'") + failures = run_system_checks(ingested) + assert any("orphan" in f for f in failures) + + +def test_enum_domain_violation_is_caught(ingested): + ingested.execute("UPDATE wh_orders SET status = 'invalid' WHERE order_id = 'o1'") + failures = run_system_checks(ingested) + assert any("enum domain" in f for f in failures) + + +def test_illegal_status_transition_is_caught(conn): + from ingestion.cdc import CDCCapture + from ingestion.pipeline import run_ingestion + from ingestion.seed import BASE_TS + + cap = CDCCapture() + cap.insert( + "customers", + "c1", + { + "customer_id": "c1", + "name": "A", + "email": "a@x.com", + "status": "active", + "created_at": BASE_TS, + "updated_at": BASE_TS, + }, + ) + order = { + "order_id": "o1", + "customer_id": "c1", + "status": "pending", + "total": 0.00, + "currency": "USD", + "created_at": BASE_TS, + "updated_at": BASE_TS, + } + cap.insert("orders", "o1", order) + # Illegal: pending -> delivered (must go through paid/shipped). + cap.update("orders", "o1", {**order, "status": "delivered"}, before=order) + run_ingestion(conn, cap) + + failures = run_business_checks(conn) + assert any("illegal status transition" in f for f in failures) diff --git a/submission/ecommerce-cdc/tests/test_warehouse.py b/submission/ecommerce-cdc/tests/test_warehouse.py new file mode 100644 index 0000000..d44b791 --- /dev/null +++ b/submission/ecommerce-cdc/tests/test_warehouse.py @@ -0,0 +1,84 @@ +"""Warehouse tests: latest snapshot, soft delete, SCD2 time travel, restore.""" + +from ingestion.pipeline import run_ingestion +from warehouse.warehouse import reconstruct_as_of, restore_from_lake + + +def test_latest_snapshot_reflects_updates(ingested): + status, total = ingested.execute( + "SELECT status, total FROM wh_orders WHERE order_id = 'o1'" + ).fetchone() + assert status == "paid" # update applied + assert float(total) == 70.00 + + +def test_delete_soft_deletes_current(ingested): + deleted = ingested.execute( + "SELECT _deleted FROM wh_customers WHERE customer_id = 'c3'" + ).fetchone()[0] + assert deleted is True + + visible = ingested.execute( + "SELECT COUNT(*) FROM wh_customers WHERE NOT _deleted" + ).fetchone()[0] + assert visible == 2 # c1, c2 + + +def test_time_travel_order_status(ingested): + # At seq 9 the order was still 'pending' (update happened at seq 10). + early = reconstruct_as_of(ingested, "orders", as_of_seq=9) + assert early[0]["status"] == "pending" + + # At seq 12 the order is 'paid'. + late = reconstruct_as_of(ingested, "orders", as_of_seq=12) + assert late[0]["status"] == "paid" + + +def test_time_travel_shows_deleted_row_before_delete(ingested): + # c3 deleted at seq 12; at seq 11 it still existed. + before = {r["customer_id"] for r in reconstruct_as_of(ingested, "customers", 11)} + after = {r["customer_id"] for r in reconstruct_as_of(ingested, "customers", 12)} + assert "c3" in before + assert "c3" not in after + + +def test_restore_from_lake_rebuilds_snapshot(ingested): + restore_from_lake(ingested, target_seq=9) + status = ingested.execute( + "SELECT status FROM wh_orders WHERE order_id = 'o1'" + ).fetchone()[0] + assert status == "pending" + + # c3 not yet deleted at seq 9 -> present and not soft-deleted. + c3 = ingested.execute( + "SELECT _deleted FROM wh_customers WHERE customer_id = 'c3'" + ).fetchone() + assert c3 is not None and c3[0] is False + + +def test_scd2_history_has_two_order_versions(ingested): + versions = ingested.execute( + "SELECT COUNT(*) FROM wh_orders_history WHERE order_id = 'o1'" + ).fetchone()[0] + assert versions == 2 # pending version + paid version + + current = ingested.execute( + "SELECT status FROM wh_orders_history " "WHERE order_id = 'o1' AND _is_current" + ).fetchone()[0] + assert current == "paid" + + +def test_out_of_order_duplicate_apply_is_safe(conn, capture): + run_ingestion(conn, capture) + # Re-applying the same records must not change the snapshot. + from warehouse.warehouse import apply_cdc_records + + apply_cdc_records(conn, capture.records_since(0)) + status = conn.execute( + "SELECT status FROM wh_orders WHERE order_id = 'o1'" + ).fetchone()[0] + assert status == "paid" + versions = conn.execute( + "SELECT COUNT(*) FROM wh_orders_history WHERE order_id = 'o1'" + ).fetchone()[0] + assert versions == 2 diff --git a/submission/ecommerce-cdc/validation/__init__.py b/submission/ecommerce-cdc/validation/__init__.py new file mode 100644 index 0000000..538a3f8 --- /dev/null +++ b/submission/ecommerce-cdc/validation/__init__.py @@ -0,0 +1 @@ +"""Validation layer: system + business parity checks against warehouse outputs.""" diff --git a/submission/ecommerce-cdc/validation/checks.py b/submission/ecommerce-cdc/validation/checks.py new file mode 100644 index 0000000..dbc5c5f --- /dev/null +++ b/submission/ecommerce-cdc/validation/checks.py @@ -0,0 +1,174 @@ +""" +Validation parity layer. + +Re-asserts the source's system and business rules against the warehouse outputs so +downstream consumers are protected even when the source is wrong. Status-transition +legality is validated from the ordered lake history (the durable record of change). + +Each function returns a list of human-readable failure messages; an empty list means +the warehouse is consistent with the source contract. +""" + +from __future__ import annotations + +from collections import defaultdict +from typing import Dict, List + +import duckdb + +from lake.lake import read_events +from source.models import ( + ORDER_STATUS_TRANSITIONS, + PAYMENT_STATUS_TRANSITIONS, + SCHEMA_CONTRACT, +) +from warehouse.warehouse import current_table + + +def run_system_checks(conn: duckdb.DuckDBPyConnection) -> List[str]: + failures: List[str] = [] + + for table, contract in SCHEMA_CONTRACT.items(): + wh = current_table(table) + pk = contract.primary_key + + total = conn.execute(f"SELECT COUNT(*) FROM {wh}").fetchone()[0] + distinct = conn.execute(f"SELECT COUNT(DISTINCT {pk}) FROM {wh}").fetchone()[0] + if total != distinct: + failures.append( + f"{wh}: PK not unique ({total} rows, {distinct} distinct {pk})" + ) + + for col in contract.columns: + if col.nullable: + continue + nulls = conn.execute( + f"SELECT COUNT(*) FROM {wh} WHERE NOT _deleted AND {col.name} IS NULL" + ).fetchone()[0] + if nulls: + failures.append(f"{wh}.{col.name}: {nulls} NULL value(s)") + + for col_name, domain in contract.enum_domains.items(): + allowed = ", ".join(f"'{v}'" for v in domain) + bad = conn.execute( + f"SELECT COUNT(*) FROM {wh} " + f"WHERE NOT _deleted AND {col_name} NOT IN ({allowed})" + ).fetchone()[0] + if bad: + failures.append(f"{wh}.{col_name}: {bad} row(s) outside enum domain") + + for fk_col, (parent, parent_col) in contract.foreign_keys.items(): + parent_wh = current_table(parent) + orphans = conn.execute( + f""" + SELECT COUNT(*) FROM {wh} c + WHERE NOT c._deleted AND c.{fk_col} IS NOT NULL + AND NOT EXISTS ( + SELECT 1 FROM {parent_wh} p + WHERE p.{parent_col} = c.{fk_col} AND NOT p._deleted + ) + """ + ).fetchone()[0] + if orphans: + failures.append( + f"{wh}.{fk_col}: {orphans} orphan(s) missing {parent_wh}.{parent_col}" + ) + + return failures + + +def run_business_checks(conn: duckdb.DuckDBPyConnection) -> List[str]: + failures: List[str] = [] + + bad_line = conn.execute( + "SELECT COUNT(*) FROM wh_order_items " + "WHERE NOT _deleted AND line_total <> quantity * unit_price" + ).fetchone()[0] + if bad_line: + failures.append( + f"wh_order_items: {bad_line} row(s) where line_total != quantity*unit_price" + ) + + mismatched = conn.execute( + """ + SELECT COUNT(*) FROM ( + SELECT o.order_id, o.total, + COALESCE(SUM(i.line_total), 0) AS items_total + FROM wh_orders o + LEFT JOIN wh_order_items i + ON i.order_id = o.order_id AND NOT i._deleted + WHERE NOT o._deleted + GROUP BY o.order_id, o.total + HAVING o.total <> COALESCE(SUM(i.line_total), 0) + ) + """ + ).fetchone()[0] + if mismatched: + failures.append( + f"wh_orders: {mismatched} order(s) where total != sum(order_items.line_total)" + ) + + neg_amount = conn.execute( + "SELECT COUNT(*) FROM wh_payments WHERE NOT _deleted AND amount < 0" + ).fetchone()[0] + if neg_amount: + failures.append(f"wh_payments: {neg_amount} row(s) with negative amount") + + bad_price = conn.execute( + "SELECT COUNT(*) FROM wh_products WHERE NOT _deleted AND price <= 0" + ).fetchone()[0] + if bad_price: + failures.append(f"wh_products: {bad_price} row(s) with non-positive price") + + bad_qty = conn.execute( + "SELECT COUNT(*) FROM wh_order_items WHERE NOT _deleted AND quantity <= 0" + ).fetchone()[0] + if bad_qty: + failures.append(f"wh_order_items: {bad_qty} row(s) with non-positive quantity") + + bad_settle = conn.execute( + "SELECT COUNT(*) FROM wh_payments " + "WHERE NOT _deleted AND settled_at IS NOT NULL AND settled_at < created_at" + ).fetchone()[0] + if bad_settle: + failures.append( + f"wh_payments: {bad_settle} row(s) where settled_at < created_at" + ) + + failures.extend( + _status_transition_failures(conn, "orders", ORDER_STATUS_TRANSITIONS) + ) + failures.extend( + _status_transition_failures(conn, "payments", PAYMENT_STATUS_TRANSITIONS) + ) + return failures + + +def run_all_checks(conn: duckdb.DuckDBPyConnection) -> List[str]: + return run_system_checks(conn) + run_business_checks(conn) + + +def _status_transition_failures( + conn: duckdb.DuckDBPyConnection, + table: str, + transitions: Dict[str, tuple], +) -> List[str]: + failures: List[str] = [] + histories: Dict[str, List[str]] = defaultdict(list) + + for event in read_events(conn): + if event["table_name"] != table or event["operation"] == "delete": + continue + status = event["data"].get("status") + if status is not None: + histories[event["primary_key"]].append(status) + + for pk, statuses in histories.items(): + for prev, curr in zip(statuses, statuses[1:]): + if prev == curr: + continue + if curr not in transitions.get(prev, ()): # illegal transition + failures.append( + f"{table} {pk}: illegal status transition {prev} -> {curr}" + ) + return failures diff --git a/submission/ecommerce-cdc/warehouse/__init__.py b/submission/ecommerce-cdc/warehouse/__init__.py new file mode 100644 index 0000000..90e9552 --- /dev/null +++ b/submission/ecommerce-cdc/warehouse/__init__.py @@ -0,0 +1 @@ +"""Warehouse layer: current snapshot, SCD2 history, restore / time travel.""" diff --git a/submission/ecommerce-cdc/warehouse/warehouse.py b/submission/ecommerce-cdc/warehouse/warehouse.py new file mode 100644 index 0000000..8b7fd1e --- /dev/null +++ b/submission/ecommerce-cdc/warehouse/warehouse.py @@ -0,0 +1,239 @@ +""" +Warehouse layer — current-state snapshot + SCD2 history built from CDC events. + +For every source table the warehouse keeps two coordinated models: + +1. ``wh_
`` current snapshot (latest state, soft-delete via ``_deleted``) +2. ``wh_
_history`` SCD2 versions (``_valid_from_seq`` / ``_valid_to_seq`` / + ``_is_current``) enabling point-in-time reconstruction + +Both are derived from the same sequence-ordered change stream, so the snapshot can +never silently drift from history. + +Restore / time travel: +- ``reconstruct_as_of`` returns table state at a given sequence from the SCD2 history. +- ``restore_from_lake`` rebuilds the current snapshot deterministically by replaying + the durable lake up to a target sequence. + +Production analogue: BigQuery/Snowflake current-state tables refreshed by a streaming +merge, with SCD2 history tables alongside for temporal queries. +""" + +from __future__ import annotations + +from typing import Any, Dict, List, Optional + +import duckdb + +from ingestion.cdc import CDCRecord +from lake.lake import read_events +from source.models import SCHEMA_CONTRACT + + +def current_table(source_table: str) -> str: + return f"wh_{source_table}" + + +def history_table(source_table: str) -> str: + return f"wh_{source_table}_history" + + +def create_warehouse_tables(conn: duckdb.DuckDBPyConnection) -> None: + """Create current + SCD2 history tables for every contracted source table.""" + for table, contract in SCHEMA_CONTRACT.items(): + col_defs = ",\n ".join( + f"{c.name} {c.dtype}" for c in contract.columns + ) + + conn.execute( + f""" + CREATE TABLE IF NOT EXISTS {current_table(table)} ( + {col_defs}, + _cdc_seq INTEGER NOT NULL, + _deleted BOOLEAN NOT NULL DEFAULT false, + PRIMARY KEY ({contract.primary_key}) + ) + """ + ) + + conn.execute( + f""" + CREATE TABLE IF NOT EXISTS {history_table(table)} ( + {col_defs}, + _valid_from_seq INTEGER NOT NULL, + _valid_to_seq INTEGER, + _is_current BOOLEAN NOT NULL DEFAULT true, + _operation VARCHAR NOT NULL, + valid_from TIMESTAMP NOT NULL, + valid_to TIMESTAMP, + PRIMARY KEY ({contract.primary_key}, _valid_from_seq) + ) + """ + ) + + +def apply_cdc_records( + conn: duckdb.DuckDBPyConnection, records: List[CDCRecord] +) -> None: + """ + Apply CDC records (in sequence order) to current snapshot + SCD2 history. + + Idempotent / out-of-order safe: a record is skipped if the row's current + ``_cdc_seq`` is already >= this record's sequence. + """ + for record in sorted(records, key=lambda r: r.sequence): + contract = SCHEMA_CONTRACT.get(record.table) + if contract is None: + continue + + pk_col = contract.primary_key + pk_val = record.primary_key + existing_seq = _current_seq(conn, record.table, pk_col, pk_val) + if existing_seq is not None and existing_seq >= record.sequence: + continue # duplicate or stale event + + captured_at = record.captured_at + if record.operation == "delete": + _close_open_version( + conn, record.table, pk_col, pk_val, record.sequence, captured_at + ) + _soft_delete_current(conn, record.table, pk_col, pk_val, record.sequence) + else: + data = _filter_columns(contract, record.image) + _close_open_version( + conn, record.table, pk_col, pk_val, record.sequence, captured_at + ) + _open_history_version( + conn, + record.table, + contract, + data, + record.operation, + record.sequence, + captured_at, + ) + _upsert_current(conn, record.table, contract, data, record.sequence) + + +# -- reconstruction / restore ------------------------------------------------- + + +def reconstruct_as_of( + conn: duckdb.DuckDBPyConnection, source_table: str, as_of_seq: int +) -> List[Dict[str, Any]]: + """Return rows of ``source_table`` as they existed at ``as_of_seq`` (time travel).""" + contract = SCHEMA_CONTRACT[source_table] + cols = ", ".join(contract.column_names) + rows = conn.execute( + f""" + SELECT {cols} + FROM {history_table(source_table)} + WHERE _valid_from_seq <= ? + AND (_valid_to_seq IS NULL OR _valid_to_seq > ?) + ORDER BY {contract.primary_key} + """, + [as_of_seq, as_of_seq], + ).fetchall() + return [dict(zip(contract.column_names, row)) for row in rows] + + +def restore_from_lake( + conn: duckdb.DuckDBPyConnection, target_seq: Optional[int] = None +) -> None: + """ + Rebuild the current snapshot from the durable lake, replaying up to ``target_seq``. + + This is the operational rollback path: it proves the warehouse can be reconstructed + purely from lake history. + """ + for table in SCHEMA_CONTRACT: + conn.execute(f"DELETE FROM {current_table(table)}") + + for event in read_events(conn, target_seq): + contract = SCHEMA_CONTRACT.get(event["table_name"]) + if contract is None: + continue + pk_col = contract.primary_key + pk_val = event["primary_key"] + seq = event["sequence"] + if event["operation"] == "delete": + existing = _current_seq(conn, event["table_name"], pk_col, pk_val) + if existing is not None: + _soft_delete_current(conn, event["table_name"], pk_col, pk_val, seq) + else: + data = _filter_columns(contract, event["data"]) + _upsert_current(conn, event["table_name"], contract, data, seq) + + +# -- internal helpers --------------------------------------------------------- + + +def _current_seq( + conn: duckdb.DuckDBPyConnection, table: str, pk_col: str, pk_val: str +) -> Optional[int]: + row = conn.execute( + f"SELECT _cdc_seq FROM {current_table(table)} WHERE {pk_col} = ?", + [pk_val], + ).fetchone() + return int(row[0]) if row else None + + +def _filter_columns(contract, image: Dict[str, Any]) -> Dict[str, Any]: + return {c.name: image.get(c.name) for c in contract.columns if c.name in image} + + +def _upsert_current(conn, table, contract, data: Dict[str, Any], seq: int) -> None: + cols = list(data.keys()) + ["_cdc_seq", "_deleted"] + vals = list(data.values()) + [seq, False] + placeholders = ", ".join(["?"] * len(vals)) + updates = ", ".join(f"{c} = excluded.{c}" for c in cols) + conn.execute( + f""" + INSERT INTO {current_table(table)} ({", ".join(cols)}) + VALUES ({placeholders}) + ON CONFLICT ({contract.primary_key}) DO UPDATE SET {updates} + """, + vals, + ) + + +def _soft_delete_current(conn, table, pk_col, pk_val, seq: int) -> None: + conn.execute( + f"UPDATE {current_table(table)} SET _deleted = true, _cdc_seq = ? " + f"WHERE {pk_col} = ?", + [seq, pk_val], + ) + + +def _close_open_version(conn, table, pk_col, pk_val, seq: int, captured_at) -> None: + conn.execute( + f""" + UPDATE {history_table(table)} + SET _valid_to_seq = ?, valid_to = ?, _is_current = false + WHERE {pk_col} = ? AND _is_current = true + """, + [seq, captured_at, pk_val], + ) + + +def _open_history_version( + conn, table, contract, data: Dict[str, Any], operation: str, seq: int, captured_at +) -> None: + cols = list(data.keys()) + [ + "_valid_from_seq", + "_valid_to_seq", + "_is_current", + "_operation", + "valid_from", + "valid_to", + ] + vals = list(data.values()) + [seq, None, True, operation, captured_at, None] + placeholders = ", ".join(["?"] * len(vals)) + conn.execute( + f""" + INSERT INTO {history_table(table)} ({", ".join(cols)}) + VALUES ({placeholders}) + ON CONFLICT ({contract.primary_key}, _valid_from_seq) DO NOTHING + """, + vals, + )