Implementation Plan — Workstream 1: One-off Warehouse Rebuild → Iceberg

Status	Draft — planning
Parent RFC	Data Warehouse v2
Issue	GH #8963
Scope	The one-off, hand-run rebuild that makes prod 100% healthy. Pulumi/IaC adoption is a separate, non-blocking workstream (2).

AWS facts below were re-verified live (--profile readonly, account 111776553558, us-west-2) at planning time. Re-verify the volatile ones (DMS status, newest raw partition, drift counts) immediately before execution.

Why two workstreams

The backfill and cutover genuinely run once — IaC buys nothing for a run-once job. The steady-state runtime (ongoing ingest, schedules, Iceberg tables, IAM, alarms) lives forever across 5 envs and must be reproducible — that belongs in Pulumi (workstream 2). This plan does the urgent one-off by hand so prod stops being broken; it does not wait on Pulumi. The one piece that straddles the seam — keeping prod fresh after backfill — is included here as a hand-run job and re-homed into Pulumi later.

Verified current state (the diagnosis)

• Pipeline: Postgres → DMS CDC (cdc-duplousprodv1-task, running) → S3 raw s3://onramp-datawarehouse-cdc/duplousprodv1-read-1/public/<table>/<YYYY>/<MM>/<DD>/ (+ a LOAD0000000N.parquet full-load file at each table root) → Glue crawler duplousprodv1 → Glue DB onramp_prod_us → Athena workgroup onramp_webapp_prod (engine v3).
• Raw bronze is healthy and current — newest or_projects partition is 2026/06; total bucket ≈ 424 GB.
• Only the catalog is broken. The crawler (CombineCompatibleSchemas + DEPRECATE_IN_DATABASE) forks a table on every column change. onramp_prod_us.or_projects is DEPRECATED_BY_CRAWLER, frozen at UpdateTime 2026-03-06, 5,249,708 records, 71,930 objects (the small-files problem).
• Scope = 178 source tables captured to raw (122 or_* app tables + 56 others), all under the DMS rule include public.% minus 10 excludes (or_documents, or_esign_templates, or_signers, or_platform_activity, or_vendors_filtered, or_analytics_execution, or_analytics_measurement, announcement_user_dismissed, announcements, or_tasks_html_entities_backup).
• Catalog cruft: 792 Glue tables = ~520 non-drift entries (178 live base tables + stale pre-crawl entries) + 272 drift tables named YYYY / YYYY_<schemahash>. The drift tables are the crawler's per-era schema forks. All 272 map cleanly to a base table via S3 location (spike, zero ambiguous), and every fork is a sub-prefix of its base's one raw root — so the backfill reads the raw root directly and subsumes them; it does not union the fork Glue tables.

Spike + dry-run validation (2026-06-09)

Two background agents validated the plan against live AWS + the dev model. Results below (the raw spike/dry-run artifacts were removed once the rebuild shipped and the warehouse went live).

Dry-run PASSED end-to-end on real AWS (2026-06-09). The full or_projects chain ran against prod raw: 7,389,191 raw == 7,389,191 Iceberg rows, 0 quarantine, reconciled; the windowed-CTE oracle hit exact contract-column parity vs the deprecated table for vendors 0/328/294/255/380/205 (watermark ≤ 2026-03-06). ~22 min on Glue G.2X×10 — the sizing baseline for the 171-table run. Notable: the job must run on Glue 5.0 (Spark 3.5) to emit TIMESTAMP_NTZ (naive, matching the deprecated table); Glue 4.0 / Spark 3.3 lacks it.

Confirmed / de-risked:

• Quarantine risk is structurally ~zero. The crawler forked tables only on column add/drop, never a type change — every shared column carries an identical type across all era forks. So backfill coercion is just NULL-fill-for-missing; expected contract-column quarantine = 0.
• No fork-union needed. Reading each base's raw root recursively with Spark mergeSchema=true covers all history (verified end-to-end on or_projects). This replaces the per-table UNION the plan originally implied — a real simplification of the backfill generator.
• Drift→base mapping is generatable from drift_table_map.json (all 167 owning bases ∈ the 178).
• Type map has no unmapped types.

Corrections folded into the phases below:

• Partition rule matches both vendor AND vendor_id (column name is inconsistent across tables) → 83 vendor-partitioned, 95 unpartitioned.
• Datetime typing refined. Only the DMS op-metadata timestamp column is string; the real datetime columns (created_dts, *_dts) are stored timestamp in the deprecated canonical table, and value is decimal(12,2). The Iceberg target must match these or the oracle output differs — the plan's blanket "timestamp→string" was too broad and applies only to the DMS metadata column.

Resolved — the 15 dev-schemaless tables (jeffg, 2026-06-09): all 15 are captured by DMS and present in raw parquet; they were missing only from the dev pg the spike read schemas from. All 15 confirmed present in prod pg, so the schema authority exists — read their target schema from prod, not dev. Split:

• Rebuild (8) — real/legacy tables: custom_field, custom_field_location_category, custom_field_location_type, custom_field_tag, custom_field_type, migration_notifications, or_module_rules, vendor_exports.
• Archive-only, no Iceberg table (7) — one-shot migration backup snapshots, left in raw: oei_4872_cleanup_backup, oei_4872_dedupe_backup, or_modules_html_entities_backup, or_projects_description_html_entities_backup, or_projects_name_html_entities_backup, or_projects_name_onramp_4596_backup, or_task_steps_html_entities_backup.

→ Rebuild scope is 171 tables (178 captured − 7 archive-only). Phase 1 pulls the current-model schema for the 8 from prod pg.

Open decision (need jeffg):

1. Prod-schema reconfirm is mandatory before execution — all current-model schemas came from local dev pg (prod unreachable from the agent). Recheck the 19 numeric columns' precision/scale; confirm dropping or_resources.search_vector_generated (tsvector) and the or_tasks.roles (int[]) parquet rendering. (The 8 rebuilt tables above are schema-from-prod by definition.)

Infra gaps surfaced by the dry-run (not blockers, but pre-reqs):

• No silver bucket exists in the account (only raw + athena-results). One must be created; SILVER_BUCKET is a required operator var.
• Workgroup onramp_webapp_prod reports EngineVersion = AUTO, not pinned v3 — confirm AUTO resolves to an Iceberg-V2-capable engine before relying on it.

Definition of done

1. All 171 in-scope tables are clean Iceberg V2 tables in onramp_prod_us, schema = current Postgres model, all history preserved, zero drift forks. (The 7 backup-snapshot tables stay raw-archive-only — no Iceberg table.)
2. Both consumer surfaces — get_project_pipeline_analytics and core-insights/pipeline-flow-analysis — return output identical to a known-good baseline.
3. Crawler duplousprodv1 deleted → drift structurally impossible.
4. Prod stays fresh (hand-run append job running).
5. No app deploy. Per-table rollback available.

The spine: consistency model

Everything hinges on a per-table watermark T (a frozen S3-object timestamp):

• Backfill covers all raw history up to T.
• The freshness job covers T → now, append-only, at-least-once.
• Cutover happens after backfill + oracle pass.
• The seam between backfill-end and freshness-start is the #1 correctness risk. Overlap is fine (the consumer's windowed CTE de-duplicates and is dup-tolerant); a gap is not. Always err toward overlap.

---

Phase 0 — Audit & freeze inputs (read-only)

1. Authoritative table manifest — the 178 raw public/<table>/ folders, cross-checked against the DMS task include public.% minus the 10 excludes and against Postgres information_schema.
2. Per-table fork map — drift_table_map.json from the spike (base → forks → S3 prefixes). Informational / cleanup-targeting; the backfill reads the raw root directly rather than these.
3. Catalog snapshot — export every Glue table DDL (rollback artifact + per-era parquet schemas).
4. Per-table watermark T — newest raw object timestamp at freeze time.
5. Confirm DMS health — prod task running, raw still landing.

Output: frozen manifest (tables + fork map + raw prefixes + per-table T).

Phase 1 — Schema authority & type map (core deliverable)

The two-schema mechanic, per table:

• Read schema (over raw parquet) = the DMS parquet physical schema; each era's drift table already carries it. Drift happens here.
• Target schema (Iceberg) = the current Postgres model; cast read → target on write.

Tasks:

1. Current-model schema per table from Postgres information_schema.columns (live authority; ORM models as cross-check).
2. pg → Iceberg type map — the real work. Explicit rules for: numeric(p,s) → decimal(p,s); real datetime columns (created_dts, *_dts) → timestamp (match the deprecated canonical table for oracle parity); the DMS op-metadata timestamp column → string (the consumer casts it in-query); jsonb/json/uuid/enums → string; bool → boolean; bytea; arrays. (Validated: no unmapped types.)
3. Type-history report — per column, how type/presence changed across forks. Drives coercion rules and flags genuinely-incompatible retypes (those rows quarantine, never silently coerce). This report sizes the quarantine risk before we commit to the rebuild.
4. Partition decision per table — has a vendor or vendor_id column? → identity-partition by it. Else unpartitioned. (Validated split: 83 partitioned, 95 unpartitioned. vendor = 0 is the real OnRamp self-tenant — never filtered anywhere.)

Phase 2 — Backfill engine (one Glue PySpark job, hand-created)

Athena INSERT…SELECT is the wrong tool at 424 GB (30-min DML timeout fights you). One Glue PySpark job, looped over the manifest. Per table:

• Read all raw for the table with a single recursive read of the raw root + Spark mergeSchema=true — this subsumes the full-load file(s), the date-partitioned CDC, and every era fork (all are sub-prefixes of the one root; verified on or_projects). No fork-table union. (IncludeOpForFullLoad means full-load rows carry the Op column, so reads are uniform.)
• Align to the target schema: select model columns, NULL-fill missing, cast per the type map. Unparseable values → quarantine side table + counts (never drop, never fail the whole table).
• Write Iceberg V2 to a staging silver location (not the canonical name yet), born-compacted (target 128–256 MB files), partitioned by vendor where present.

Operational: hand-create the job + IAM role (read raw, write silver, Glue catalog). Size workers from a single-table dry run before the full run. Job is idempotent / re-runnable.

Phase 3 — Oracle gate (per table, blocks cutover)

• Deduped-raw set-equality — dedup raw by (id, timestamp, op, changed-col tuple); assert set-equality on contract columns vs the Iceberg table, per vendor, all history.
• Quarantine count must be 0 for contract columns or stop.
• For consumer tables (or_projects first): run the actual windowed CTE (VendorProjectPipelineHelper.get_project_history, app/api/vendor/helpers.py:757) old-vs-new, per vendor (incl. vendor = 0), bounded by a watermark date (the deprecated table is frozen at 2026-03-06, so compare apples-to-apples up to that bound). Assert both change-event row count and a contract-column checksum (checksum catches value drift when counts match). Both HTTP surfaces funnel through this one CTE, so matching it proves both.
• Chunk oracle queries by vendor/date to stay under the Athena timeout.

Phase 4 — Cutover (atomic catalog repoint, per table)

1. Rehearse in dev first; export DDL first.
2. Per table: back up get-table JSON → update-table repoint canonical name → Iceberg metadata (CAS). Fallback: Athena ALTER TABLE … RENAME.
3. Order: flip or_projects first → validate live on both surfaces → then bulk-flip the rest.
4. No app deploy — the Athena DB map already resolves prod → onramp_prod_us.

Phase 5 — Kill the drift engine + freshness (in scope, hand-run)

• Delete crawler duplousprodv1 once Iceberg is canonical → drift can no longer occur.
• Stand up a minimal hand-run freshness job: scheduled, drains new raw per table by watermark → INSERT INTO the Iceberg table (at-least-once; consumer dup-tolerant). Plus scheduled OPTIMIZE / AWS-managed compaction for the small-files problem.
• Seam note: this freshness job is steady-state and will be re-homed into Pulumi (workstream 2). It lives here only so prod doesn't re-stale the moment backfill ends.

Phase 6 — Cleanup & rollback readiness

• Rollback = per-table catalog backups → re-point if a post-cutover oracle regresses.
• Raw bronze stays immutable (the truth; everything is rebuildable from it).
• Defer deleting the 272 drift tables — first inventory Superset metadata + Athena/CloudTrail (≥ 90 days) for any direct dependency. Off the critical path.

---

Defaulted decisions

Decision	Default	Rationale
Schema authority	Postgres information_schema	Live model, trivially queryable; ORM as cross-check
Backfill engine	Glue PySpark (not Athena-only)	424 GB at warehouse scale; Athena timeout fights you
Iceberg version	V2	Athena/Trino cannot read V3
Cutover order	or_projects first, then bulk	De-risk on the one consumed table before the herd
Freshness job	In scope, hand-run	Prod re-stales immediately otherwise; re-home into Pulumi later
Drift-table deletion	Deferred	Dependency inventory needed; off critical path

Top risks

• Backfill/freshness seam gap → explicit per-table watermark + deliberate overlap.
• Genuinely incompatible retypes → quarantine volume could be non-trivial; the Phase-1 type-history report sizes it before committing.
• Glue cost/time on 424 GB → single-table dry run sizes workers first.
• Stale memory / drift recount → re-verify DMS status, newest partition, and the 272/178 counts immediately before execution.

Explicitly deferred to Workstream 2 (Pulumi / steady-state)

• IaC for all persistent resources (ingest job, schedules, Iceberg tables, IAM, buckets, alarms), importing prod and recreating the lower envs.
• Multi-env rollout (stage / demo / gdpr-prod / dev rebuild).
• Observability suite (freshness / Glue-fail / DMS-FATAL + source-side replication-slot alarms).
• GDPR whole-vendor erasure CLI.
• Deletion of the 272 drift tables after the dependency-inventory window.