Runbook — Fase 4: flip WRITERS to iceberg-native (massive ingest on S3)

Stage 1 · Fase 4. Flip dataset-row writers from Postgres to the Iceberg lakehouse, so source ingest lands directly on S3 via the chunked add_files writer (RAM bound by chunk) instead of dataset_rows + async mirror. This is an operational procedure (DB flag updates against a live ml-runner) — the iceberg-native write path + the G1–G4 concurrency guards already ship in code.

This is the write-side mirror of lakehouse-fase1-read-flip.md. The mechanics (registry defaults → runtime overrides, scope precedence, ~10s flag-cache TTL) are the same; the safety model is different — a write PRODUCES the truth, so there is no PG fallback (fail-loud) and the ordering constraint below is mandatory.

⚠️ Prerequisite #1 — READS must already serve Iceberg for the dataset

Flipping a writer to iceberg_native means new writes skip dataset_rows and the mirror poller skips that dataset. Postgres immediately goes stale for it. So every reader of that dataset must already read from Iceberg before you flip the writer — otherwise consumers read stale/empty PG.

• Do the read-flip first (lakehouse-fase1-read-flip.md for EVENTUAL readers; Fase 3/4 for STRICT) and confirm icebergIdentityReady(datasetId) + a succeeded snapshot.
• Rule of thumb: read-flip a dataset → soak → then write-flip it. Never the reverse.

Scope — writers that CAN flip (maxMode = iceberg_native)

Locked to PG (do NOT flip — maxMode=pg, the CLI refuses them): ingest.stream (streaming txn), row.edit (point DELETE has no Iceberg primitive → Karma), dataset.manifest (file-backed pointers), pipeline.output (Karma compute).

Why each flip is safe

• Fail-loud, no fallback (lib/lakehouse/write-router.ts): a native write that fails marks the dataset errored and the sync retries — it never silently split-brains by writing dataset_rows after an Iceberg attempt.
• Capability ceiling: the write-flags CLI refuses any mode above a writer's maxMode (the router would clamp it anyway).
• Concurrency guards (G1–G4): the shared catalog is serialized (G1), ingest is admission-bounded (G2), same-dataset ingests serialize (G3), and commit conflicts retry (G4). Validate them at your target concurrency with the bench (step 2) before rolling out.
• ~10s flag-cache TTL: a flip or rollback takes effect within ~10s.

Pre-requisites

1. ML_RUNNER_URL + ML_RUNNER_TOKEN in .env.local.
2. Migrations applied: identity (__row_index/__row_id) + __created_at; the target datasets identity-backfilled (icebergIdentityReady).
3. The Fase 4 writer + G1–G4 guards deployed to the live ml-runner.
4. Prerequisite #1 satisfied for the candidate datasets (reads already on Iceberg).

Procedure

1. Inspect current write-flag state

npm run lakehouse:write-flags -- list

Shows registry defaults (all writers pg) + ceilings + any runtime overrides.

2. Validate the guards at target concurrency (concurrency bench)

Before routing real traffic, confirm the guards hold and size max_concurrent_ingest:

# distinct-dataset burst — expect error=0 (G1) and 503s only if jobs > max_concurrent_ingest (G2)
dotenv -e .env.local -- tsx scripts/lakehouse-ingest-bench.ts run --jobs 32 --rows 500000 --namespace bench
# same-dataset burst — expect error=0 (G3 serialization; no replace drop-race)
dotenv -e .env.local -- tsx scripts/lakehouse-ingest-bench.ts run --jobs 16 --same-dataset --namespace bench

Gate: non-503 error must be 0. 503s are healthy backpressure (raise max_concurrent_ingest or accept Node/BullMQ retries). Purge the bench namespace after.

2b. Candidate selection (F3) — exclude pipeline-input datasets

A native-write dataset's dataset_rows goes empty, but the pipeline source CTE (lib/pipelines/sourceCte.ts) reads FROM dataset_rows WHERE dataset_id = $1. So a dataset that feeds any pipeline as INPUT is a HARD BLOCKER — it must NOT be write-flipped until sourceCte reads Iceberg (C3 / native compute). This read-only query classifies identity-ready datasets as SAFE vs BLOCKED:

with latest as (
  select distinct on (dataset_id) dataset_id, status, has_identity
  from iceberg_sync_log order by dataset_id, finished_at desc nulls last),
pin as (   -- datasets referenced as a pipeline INPUT (a `dataset` spec node)
  select distinct (node->'data'->>'datasetId')::uuid as dataset_id
  from pipelines, jsonb_array_elements(spec->'nodes') node
  where node->>'type' = 'dataset'
    and (node->'data'->>'datasetId') ~ '^[0-9a-fA-F-]{36}$')
select case when p.dataset_id is not null then 'BLOCKED (pipeline input)' else 'SAFE' end as verdict,
       d.name, d.row_count, d.id
from latest l
join datasets d on d.id = l.dataset_id
left join pin p on p.dataset_id = l.dataset_id
where l.status = 'succeeded' and l.has_identity
order by verdict, d.row_count desc nulls last;

Only flip writers for SAFE datasets. BLOCKED ones wait for the pipeline-CTE migration to Iceberg (Fase 4 native compute).

3. Flip ONE dataset, sync, verify parity

npm run lakehouse:write-flags -- set sync.full iceberg_native --dataset <uuid> --note "fase4 canary"

Trigger a sync for that dataset, then verify:

• iceberg_sync_log has a fresh succeeded row with the expected rows + snapshot;
• datasets.row_count matches the source;
• content parity PG-vs-Iceberg via the checksum gate (order-independent digest):

npm run lakehouse:canary -- --smoke --dataset <uuid>

ok → parity holds, safe to widen. MISMATCH on a caught-up dataset → STOP, roll back, investigate. (For a REPLACE flip the PG rows are what the mirror last wrote; after the write-flip PG stops updating — run the parity check on the FIRST post-flip sync, then rely on the reads being Iceberg.)

4. Expand scope: workspace → global, one writer at a time

npm run lakehouse:write-flags -- set sync.full iceberg_native --workspace <uuid> --note "fase4 ws"
npm run lakehouse:write-flags -- set sync.full iceberg_native --note "fase4 global"   # no scope = global

Most-specific scope wins (dataset > workspace > global). Roll out one writer at a time (start with sync.full, then sync.append, then sync.incremental/cdc.append); soak between steps and watch the guards.

Monitoring

• ml-runner logs: iceberg_native: <dataset> (<rows> rows → snapshot …) on success; iceberg_native write failed marks the dataset errored (fail-loud).
• iceberg_sync_log: succeeded snapshots per dataset (has_identity=true).
• Admission: 503 rate on /lakehouse/ingest (Node worker logs retries) → if sustained, raise max_concurrent_ingest.
• RAM: peak ingest RAM ≈ max_concurrent_ingest × LAKEHOUSE_INGEST_CHUNK_ROWS rows.

Rollback (instant, ~10s)

npm run lakehouse:write-flags -- set sync.full pg --dataset <uuid>   # force PG for one dataset
npm run lakehouse:write-flags -- clear sync.full                     # drop override → registry default (pg)

No deploy needed; the flag cache refreshes within ~10s. NB: rolling a writer back to pg resumes dataset_rows writes, but rows written to Iceberg while native are NOT back-filled to PG — only roll back before a dataset's consumers depend on the Iceberg-only data, or re-run a full_sync on PG to repopulate.

Residual / next

• RestCatalog / Lakekeeper cutover (LAKEHOUSE_CATALOG_TYPE=rest) removes the G1 in-process catalog lock (server-side commit serialization) and is the prerequisite for scaling the ml-runner to multiple replicas. See the Fase 4 hardening notes.
• Chunk sizing: tune LAKEHOUSE_INGEST_CHUNK_ROWS (file size) and max_concurrent_ingest (RAM/throughput) per the bench.