⚙️ 2d.1 — Dirigir el write de pipeline.output al tridente (approach)

0.5 El camino MÍNIMO (lo que 2d.1 SÍ hace)

1. Brazo iceberg: en el write dispatch (§4): las filas computadas por el SQL de pipeline → facade.write({kind:'rows', writeMode:'replace', sourceRefs}) → tridente. Reusa el patrón manual-table (write nativo post-commit de la metadata PG). El compute PG se lee simple (cursor o memoria acotada — no hace falta optimizar; Karma lo reemplaza).
2. Subir maxMode pg→iceberg_native (writer-flags.ts:109-112, defaultMode sigue pg) → flip opt-in por-dataset (canary).
3. Fix 2 bugs reales de nativo (§6): E3 recount SELECT count(*) → usar el count del CommitResult; E4 cross-join pre-flight → handle.rowCount.
4. replace-first, canary por-dataset (Tier A: join/split/transform-snapshot/1er-deploy). Sin delta nativo (solo replace — simple).

Atomicidad SIN ledger elaborado: facade.write() (→ withNativeControlPlane) YA es open→commit atómico y fail-loud para el write nativo (snapshot Iceberg all-or-nothing). La metadata PG (row_count/watermark) se estampa después, como en manual-table; si el proceso muere entre medias, un re-deploy (replace, idempotente) lo reconcilia. No construimos el write-intent ledger con estado open revivido + índice-único + barredor — eso es Karma-era.

0. Correcciones que la auditoría de 2d.1 añadió

Además: NativeWriteResult no carga skipped/preserved (solo el total rows) → los merges Tier-B calculan esos conteos en el productor (Node), no ml-runner.

1. Las cinco piezas del motor

compute PG (cursor, rollbackable)                    [PIEZA A]
   → AsyncIterable<Row> de filas de negocio SIN identidad
        → facade.write({kind:'rows', writeMode:'replace', sourceRefs})   [PIEZA C]
             → ml-runner estampa __row_index/__row_id/__created_at → snapshot Iceberg
   envuelto en el WRITE-INTENT LEDGER (open→commit, tx partida)          [PIEZA B]
        gobernado por maxMode + canary + full-rebuild-first + watermark  [PIEZA D]
   con los BORDES cerrados (recount, cross-join, checkpoint, …)          [PIEZA E]

2. PIEZA A — compute→stream (el primitivo técnico)

El SELECT de negocio YA está aislado en cada productor — solo hay que despojar el envoltorio INSERT … to_jsonb(…) , row_number():

• estrategias deploy: el CTE __final__ / __combined__ / __filtered__ (dataset-write-strategies/{snapshot-replace:137-143, append-new:82-88, append-always:59-61}.ts) vía buildOutputProjection (projection-sql.ts).
• transform: finalSelect de combineSourceWithChain (sourceCte.ts:135-150, usado en transformPaths.ts:347).
• join: compiled.sql (transformPaths.ts:764). split: matchedSql/unmatchedSql (dos streams).
• model-node: ya es windowed (readSql, model-node-materialiser.ts:204-213) — caso especial (ver §7).

Mecanismo: cursor PG server-side vía pg-query-stream (ya es dependencia; patrón de referencia con cleanup en postgresql-adapter.ts:807-889; alternativa DECLARE/FETCH en cdc/snapshot.ts:43-62). Memory-bound a 1 batch. Se ejecuta WITH ${ctes} SELECT <cols> FROM __final__ como cursor sobre el mismo FROM ${sourceTable} que ya resuelve resolvePipelineSource (P0.2). El AsyncIterable<Row> alimenta directo facade.write({rows}) (que ya lo consume vía buildNdjsonBody, memory-bound, con backpressure de red). columns = outputSchema es pass-through (misma forma que NativeWriteColumns).

Reusar: pg-query-stream, el patrón flatten() batch→row de DatasetWriter.writeDatasetIcebergNative (dataset-writer.ts:1003-1018), la proyección __final__. Construir: el adaptador cursor→AsyncIterable sobre el SELECT compilado del pipeline.

⚠️ Decisión abierta clave (tx larga): el cursor mantiene la tx PG del compute abierta mientras ml-runner ingesta a R2 (segundos–minutos). Dos opciones (ver §8):

• (a) cursor directo — una tx PG viva toda la ingesta (simple, pero retiene conexión + arriesga statement_timeout).
• (b) SELECT→temp — CREATE TEMP TABLE __out ON COMMIT DROP AS <select>, COMMIT del compute, luego stream de la temp con una tx corta de lectura. Desacopla compute-tx de ingest, pero duplica escritura.

3. PIEZA B — write-intent ledger + atomicidad

Revive el estado open muerto de pipeline_build_transactions (CHECK ya lo admite; hoy el pipeline inserta committed inline en dataset-output-materialiser.ts:348 y transformPaths.ts:326). Máquina de estados (clon del withNativeControlPlane, iceberg-native-write.ts:125-231, adaptado a tx partida):

INTENT   [tx-PG-A corta]  INSERT pipeline_build_transactions status='open',
                          transaction_id=τ (pre-minted), row_count=0
COMPUTE  [rollbackable]   cursor PG → AsyncIterable<Row> (NO INSERT dataset_rows)
WRITE    [externo]        facade.write({replace}) → snapshot Iceberg all-or-nothing
COMMIT   [tx-PG-B corta]  UPDATE status='committed', row_count=N (del CommitResult),
                          committed_at=τ_real; + source_watermark = τ_real (SOLO AQUÍ);
                          + datasets.row_count/status='active'; + runExpectations

Dónde se parte la tx única de hoy (dataset-output-materialiser.ts — todo hoy en una tx por output, deploy-worker.ts:607-680):

• Queda en tx-PG-A (compute, rollbackable): el compute que produce el stream. En replace-first NO hay DELETE/INSERT de dataset_rows. El UPDATE datasets.schema (:271) se difiere al COMMIT (un compute fallido no debe adelantar schema).
• Va a tx-PG-B (post-write): pipeline_build_transactions→committed, source_watermark, row_count/status, expectations. El reconcileDatasetSchema (:380) desaparece (replace reescribe limpio; ver §6).

Serialización (refina P0.4, E2): el advisory-lock xact-scoped NO cubre el ciclo largo. Recomendado: unique partial index pipeline_build_transactions(dataset_id) WHERE status='open' → dos INTENT sobre el mismo dataset colisionan; el 2º espera/reintenta. Abarca INTENT→COMMIT naturalmente y se libera al marcar committed/aborted. El advisory-lock actual se conserva solo dentro de la tx-PG-B corta si hace falta ordenar los UPDATEs finales.

Idempotencia + recovery: replace es idempotente (ml-runner posee row_index → mismo snapshot lógico) → retry seguro tras commit parcial (write OK, tx-PG-B falló). append-desde-max NO lo es → replace-first. Construir un barredor (no existe): job que marca open AND created_at < now()-T → aborted (benigno: replace lo regenera; con el índice-único, también libera la serialización).

Reusar: los pasos commit+stats de withNativeControlPlane (casi 1:1 con la tx-PG-B); dataset_transactions_all + windowAllowsDelta/watermarkLowerBound (incremental.ts) tal cual. Construir: el estado open separado, el barredor, committedAt de vuelta (E1), el índice-único de serialización.

4. PIEZA C — el brazo iceberg: + mapeo por estrategia

El contrato de estrategia se extiende con un productor de filas (compute sin write):

interface WriteStrategy {
  apply(ctx): Promise<WriteResult>;        // PG (hoy, intacto)
  computeRows(ctx): { rows: AsyncIterable<Row>; writeMode; columns };  // NUEVO
}

El despacho (dataset-output-materialiser.ts:330, transformPaths.ts:362, model-node-materialiser.ts:249) gana el brazo:

sink.run({
  postgres: () => strategy.apply(ctx),                       // intacto
  iceberg: async () => {
    const { rows, writeMode, columns } = strategy.computeRows(ctx);   // cursor
    const commit = await loadItemForConsumption({datasetId: producedDatasetId}, 'write',
        {writerId:'pipeline.output', workspaceId}, {writeMode})
      .then(h => h.write({kind:'rows', columns, rows, writeMode,
        sourceRefs:[{datasetId: output.sourceDatasetId}],
        metadata:{service:'pipeline', origin:'pipeline_output'}}));
    return commitToWriteResult(commit, ctx);
  },
});

sourceRefs propaga la gobernanza (Fase D, mergeUpstreamProvenance) → reemplaza el INSERT pipeline_build_transactions manual. Plantilla E2E de referencia: el manual-table route (app/api/pipelines/[id]/manual-table/route.ts:196-221, commit PG de metadata → write nativo post-COMMIT).

Ranking replace-first (Tier A = primer canary, mapean limpio; Tier B = 2d.3, leen el target):

CommitResult→WriteResult: Tier A → rowsWritten=commit.rows, skipped=0, preserved=0. Tier B → el merge en Node calcula skipped/preserved. upsert de PyIceberg ≠ snapshot_replace preservante (M6) → Tier B se hace leyendo el target por facade + merge en el productor (traducir __new_deduped__/__preserved__/__filtered__ a Node) + write({replace}) del combinado (idempotente).

Confirmar (construir): la semántica del conteo append de ml-runner (¿total o batch?).

5. PIEZA D — maxMode, full-rebuild-first, watermark, canary

Subir el techo (writer-flags.ts:109-112): maxMode:'pg'→'iceberg_native', defaultMode sigue pg → flip opt-in por-dataset. Sin el brazo iceberg, el primer flag revienta el deploy (fail-loud write-router.ts:189-192) — por eso el brazo (§4) es prerrequisito de subir el techo.

Full-rebuild-first (E-M3, decidido): clampar readMode→snapshot cuando el sink es iceberg, en DOS sitios (dataset-output-materialiser.ts:290-307 + transformPaths.ts:293-310). Requiere mover resolveDatasetRowSink ARRIBA del bloque delta (hoy se resuelve en :327/:361, después). Helper compartido sinkForcesFullRebuild(sink) en incremental.ts (la regla "iceberg ⇒ snapshot" en un solo lugar; NO tocar windowAllowsDelta, que es pura y sirve a downstreams PG legítimos). Evita el mixed-mode (deltaPredicate sobre drenado lee 0 en silencio).

Watermark (E1): hoy high=buildStart wall-clock (dataset-output-materialiser.ts:290,302,355). Añadir committedAt al return de NativeWriteResult (iceberg-native-write.ts:217-223 + interfaz :44-46); el productor usa high = commit.committedAt al escribir el watermark en la tx-PG-B. Así la ventana (low, high] del downstream y el sello de la txn son el MISMO reloj → sin skew. Importa incluso con full-rebuild-first (downstreams que lean este output en delta).

Canary por-dataset: extender CUTOVER_WRITERS (lakehouse-cutover.ts:44) — segmentado (pipeline.output no lo tocan sync.*/ingest.api; no mezclar en la misma lista). Disparador del flip = re-deploy (no connector sync). Secuencia: check → set pipeline.output iceberg_native --dataset <ds> → re-deploy → verify (checksum) → canary --smoke. Primero hojas del DAG, replace-pura, sin downstream, ≤1M filas. Revert = clear/set pg (≤10s TTL). NO drenar dataset_rows hasta paridad sostenida (Fase 5).

Observabilidad: loguear la decisión de sink (sink.sink/mode/clamped) en el log de estrategia (dataset-output-materialiser.ts:358-370); añadir warn al fallback NO-EXEC del read-seam (read-router.ts:206-213). El canary existente ya muestrea iceberg_sync_log status='succeeded' → captará outputs nativos de pipeline automáticamente.

6. PIEZA E — bordes + checklist de completitud

Fuera de scope confirmado: model-prepass stamping (control-plane, agnóstico al substrato), object_type write (substrato objects), model-node row_count (contador en memoria), split/join recount (cuentan el propio INSERT, no el target).

7. Secuenciación (camino MÍNIMO — el motor es Karma)

2d.1a · FIX de bordes (inert — PG hoy, cero cambio de comportamiento)   ✅ HECHO (2026-07-04)
     ├─ ✅ E3: transform-path recount decoplado — snapshot usa rowsWritten (sin leer
     │       dataset_rows, native-ready: el count vendrá del CommitResult); delta
     │       mantiene el recount (PG-only, native=full-rebuild=snapshot)
     ├─ ✅ E4: cross-join pre-flight cuenta datasets.row_count (source-agnostic) en vez
     │       de count(*) FROM dataset_rows (que da 0 para una fuente nativa → guard off)
     └─ (opcional, diferido) logging de sink + warn NO-EXEC (observabilidad)

2d.1b · EL BRAZO (inert — defaultMode sigue pg, ningún flag flipeado)   ◑ PARCIAL (2026-07-04)
     ├─ ✅ helper `lib/pipelines/native-output-write.ts` (writeOutputToTrident): corre el
     │      compute PG (read acotado en memoria, cap PIPELINE_OUTPUT_NATIVE_MAX_ROWS; Karma
     │      streamea) → facade.write({replace, sourceRefs}) → tridente. Count del CommitResult.
     ├─ ✅ brazo iceberg: en TRANSFORM (applyTransformPath): sink.run({postgres, iceberg}),
     │      computeSql = WITH ${ctes} SELECT to_jsonb(finalSelect.*) AS data; delta → throw
     │      (full-rebuild-first: nativo = snapshot).
     ├─ ✅ subir pipeline.output.maxMode pg→iceberg_native (defaultMode sigue pg → canary opt-in)
     Atomicidad interina SIN ledger nuevo: withNativeControlPlane (snapshot all-or-nothing) +
     re-apply idempotente (replace). committedAt preciso + tx partida = Karma-era.

2d.1c · APPLY-PATH COMPLETO (inert)   ✅ HECHO (2026-07-04)
     ├─ ✅ brazo iceberg en JOIN (applyJoinPath): computeSql WITH leftCte,rightCte SELECT
     │      to_jsonb(compiled.sql). Full replace (sin delta). Fuentes linchpin-routed.
     └─ ✅ brazo iceberg en SPLIT (applySplitPath, DUAL-TARGET): ambas ramas iceberg → DOS
            writes nativos (matched + unmatched, una snapshot cada uno); ambas pg → el
            statement combinado; mixto → throw (flipear el par junto).

2d.1d · DEPLOY SOURCE-RESPECTING (inert)   ◑ PARCIAL (2026-07-04)
     ├─ ✅ deploy-strategies SOURCE read → linchpin: `WriteContext.sourceTable` resuelto por
     │      `resolvePipelineSource` en el materialiser + las 3 estrategias leen
     │      `FROM ${sourceTable}` (append-always/new/snapshot-replace). Un upstream nativo se
     │      hidrata; delta fuerza PG. SQL byte-idéntico hoy (sourceTable='dataset_rows').
     └─ ⏳ el WRITE arm nativo del deploy + model-node = TIER B / contrato (2d.3+):
            · deploy: computeRows por estrategia + brazo; snapshot_replace/append_new son
              merge/append (Tier B, target-read por facade); solo pure-replace es Tier A.
            · model-node: col `__source_row_key` explícita (correlación) + brazo.
            Hasta cablearse, un flip de un output de deploy/model lanza fail-loud
            (write-router:189) — SEGURO porque defaultMode=pg.

NOTA: la suite lib/workers/pipelines/dataset-write-strategies tiene 8 fallos PRE-EXISTENTES
en main (projection-sql `::int` casts + behavior downstream), independientes de esta
migración — verificado idéntico en el commit limpio a9e72b4. El source-routing es neutral.

2d.2 · PRIMER CANARY (Tier A replace-pura, hoja del DAG, ≤1M, sin checkpoint transient)
     └─ join → transform-snapshot → split → snapshot_replace-1er-deploy
        (check→flip dataset-scoped→re-deploy→verify checksum→canary smoke; revert ≤10s)

DIFERIDO A KARMA / innecesario ahora (NO construir en 2d.1):
     ├─ write-intent ledger elaborado (estado `open` revivido + unique-index + barredor)
     │    → facade.write ya es open→commit atómico; la metadata PG sigue post-write (manual-table)
     ├─ committedAt preciso + full-rebuild-first del delta nativo (solo replace, sin delta nativo)
     ├─ cursor pg-query-stream optimizado (basta un read simple; Karma reemplaza el compute)
     └─ Tier B merge-preservante (append_new / snapshot_replace preservante) → cuando toque

Clave: el motor de compute es Karma (no llegado); 2d.1 NO lo construye. Solo dirige el write al tridente vía facade.write() (que ya existe). 2d.1a/b son inert-but-ready (verificables con typecheck+seam+vitest sin infra); 2d.2 es el primer flip real (necesita ml-runner+R2+Lakekeeper+canary). La atomicidad la da withNativeControlPlane (snapshot Iceberg all-or-nothing) + el re-deploy idempotente (replace), sin ledger nuevo.

8. Decisiones abiertas (con recomendación)

9. Anti-patrones (que este approach evita)

• ❌ Correr el write nativo dentro del SAVEPOINT del deploy → retiene conexión + lock durante la ingesta a R2.
• ❌ Confiar en el advisory-lock xact-scoped para el ciclo largo (E2) → se suelta durante el write externo.
• ❌ Avanzar el watermark antes del COMMIT del ledger, o con buildStart local (E1) → delta saltado.
• ❌ Estampar row_count desde SELECT count(*) FROM dataset_rows sobre un nativo drenado (E3) → 0.
• ❌ Flipear un output >1M fuente de otro pipeline antes del compute nativo → hard-fail en el cap 1M.
• ❌ append/merge nativo antes que replace-first → no idempotente tras commit parcial.
• ❌ Flipear un dataset fuente de checkpoint transient sin el guard (E5) → redirect→vacío silencioso.

Approach implementation-ready, fundado en 5 auditorías (2026-07-04). El motor sigue a los P0 (cerrados) y precede al primer canary (2d.2). Mantener al día conforme cada pieza aterrice.