🚰 PIEZA · INGESTION — de la fuente al bucket

Qué es

Qué NO es

• No es Spark. El cómputo del cluster no participa: la extracción es Node y la escritura es Python/PyIceberg. La ingesta se quedó en el paradigma anterior.
• No es el write-path del SQL Editor. Aquél entra por la puerta (Junction) y escribe con el motor. Éste es un carril separado, con otras piezas y otras garantías.
• ⛔ No está gobernada por la cara. Ver §«Lo que el trazado destapa».

El flujo, tal como está en el código

 ① FUENTE — el Postgres del cliente
    credencial en Index · su `metadata` lleva el destino (catálogo + esquema)
      │
 ② EXTRACCIÓN — Node
    lib/workers/polling/postgresql-polling-worker.ts   → PostgreSQLPollingWorker
    lib/workers/polling/base-polling-worker.ts         → el motor común de sondeo
    lib/workers/adapters/database/postgresql-adapter.ts → introspección + lotes
    ⭐ v7: `fetch batch → yield` — streaming. El límite en Node es UNA fila
      │
 ③ COORDENADA — Index          lib/workers/polling/dataset-writer.ts
    findOrCreateDataset() ─▶ acuña (workspace, schema_id, name)
    ⭐ Aquí, y sólo aquí, la ingesta toca gobernanza: el dataset NACE con su sitio
      │
 ④ EL DESVÍO                   resolveDatasetRowSink(dataset)
    └── 'iceberg'  ⇒ el ÚNICO camino. El destino lo decide la CAPACIDAD del
                     puerto (`write-router:102` → `mode = spec.maxMode`), no un
                     flag, y los tres del sync declaran `iceberg_native`
    🪦 'pg'        ⇒ RETIRADO el 08-13. Hoy lanza (ver §②)
      │
 ⑤ TRANSPORTE — Node → Python  lib/lakehouse/iceberg-native-write.ts
    writeIcebergNativeSnapshot  (un POST)   ·  writeIcebergNativeFanout (por particiones)
    lib/lakehouse/ingest-client.ts
      └─ POST {WAREHOUSE_WRITER_URL}/lakehouse/ingest   · NDJSON en streaming
         (meta + una fila por línea; Node nunca sostiene el dataset en memoria)
      │
 ⑥ ESCRITURA — warehouse-writer (Railway)  services/ml-runner/app/lakehouse/writer.py
    PyIceberg · build_catalog(type='rest', uri = LAKEHOUSE_REST_URI)
    ⛔ LAKEHOUSE_REST_URI = https://heroic-victory-…/catalog  →  LAKEKEEPER DIRECTO
    PyArrow → Parquet   ·   catálogo → commit del snapshot
      │
 ⑦ BYTES — Cloudflare R2 (bucket `lakehouse`, WEUR)

De dónde sale la credencial, y quién decide su destino

 ⓪ EL ALTA — app/api/integrations/[service]/connect/route.ts
    Paso 1 del asistente → `user_credentials`  (encrypted_data · workspace_id · created_by)
    Paso 2 del asistente → el DESTINO en el Warehouse (catálogo + esquema)
       └─ se guarda en `user_credentials.metadata.warehouse_destination`
          { catalogId · schemaId · catalogName · schemaName }
      │
 ③′ QUIÉN LO LEE — DatasetWriter.ensureWorkspaceId()
    lee la credencial UNA vez y se lleva de paso el destino
       └─ ensureTargetSchemaId():
            1. `warehouse_destination.schemaId`   ← lo que eligió el usuario
            2. si no hay → el `main.default` del workspace   ← decisión IMPLÍCITA

⇒ El destino lo decide el usuario en el alta, y si no lo decide, lo decide el código. No hay un tercer sitio: la coordenada se acuña en findOrCreateDataset con lo que salga de ahí.

Lo medido (2026-08-12)

user_credentials ............ 26
   con warehouse_destination .  9      ⇒ 17 fuentes (65 %) SIN destino elegido
datasets sin schema_id ......  0      ✅ la coordenada está completa

⚠️ Las 17 sin destino no están rotas: están en main.default. Pero es un destino que nadie eligió — y como el paso 2 del asistente es opcional y se persiste best-effort (un fallo sólo escribe un warn), la diferencia entre «lo eligió» y «se lo pusimos» no queda registrada en ninguna parte.

⛔ Y un bug real en el camino de respaldo

dataset-writer.ts resuelve el workspace desde la credencial; si no la hay, cae a «el workspace que este usuario posee»:

.from('workspace_members').select('workspace_id')
.eq('created_by', this.userId)      // ⛔ esa columna NO EXISTE
.eq('role', 'owner')

workspace_members es (workspace_id, user_id, role, invited_by, invited_at, joined_at) — no tiene created_by. Quien lo tiene es user_credentials, de donde probablemente se copió. El predicado correcto es user_id.

⇒ Ese respaldo no puede funcionar nunca, y el error que produce —«No credential and no owned workspace found»— manda a buscar al sitio equivocado: dice que no hay workspace cuando lo que hay es una columna mal escrita. Hoy no muerde porque todas las rutas llegan con credencial.

Cómo se descubre el esquema, y cómo se traducen los tipos

 ②′ DESCUBRIMIENTO — lib/workers/adapters/database/postgresql-adapter.ts
    information_schema.tables  ⚠️ MULTI-ESQUEMA a propósito: NO se ancla al
                                  `schema` configurado en la credencial
    + FKs en una sola consulta (table_constraints ⋈ key_column_usage ⋈
      constraint_column_usage) — con su bandera de «la introspección falló»

 ②″ TRADUCCIÓN — dos saltos y medio, y cada uno tiene dueño distinto
    ① tipo SQL de la fuente ──normalizeDbType──▶ BASE_TYPE   (el tipo lógico de Carbon)
    ② BASE_TYPE ──BASE_TYPE_TO_ICEBERG──▶ tipo Iceberg        lib/types/iceberg-types.ts
    ③ tipo Iceberg ──────────────────────▶ pyarrow            ml-runner (mecánico)

⭐ El reparto de dueños está bien pensado, y conviene decirlo: TypeScript posee el mapeo de dominio (BASE_TYPE → Iceberg) porque deriva con BASE_TYPES; ml-runner posee el mecánico (Iceberg → pyarrow) porque es la spec y no deriva. La consecuencia buscada: ml-runner es un ejecutor tonto que no sabe nada de BASE_TYPES.

Lo que se pierde por el camino

⛔ Y el espejo que sostiene un comentario

La lista de los 12 tipos JSON existe dos veces: JSON_ENCODED_BASE_TYPES en lib/types/iceberg-types.ts y en services/ml-runner/app/lakehouse/decode.py, con un # MUST mirror encima.

Hoy coinciden — verificado, 12 y 12, mismos nombres. Pero:

# los trinquetes que SÍ existen en el repo:
check:sdk-spec · check:pipeline-drift · check:iceberg-freshness · check:dataset-rows-seam
check:storage-perimeter · check:cedar · check:namespace-resolution · check:dataset-births
check:row-index-seam · check:catalog-sql · check:carbon-sql-contract · check:plan-diferencial
# ⛔ ninguno compara estas dos listas

⇒ El mismo conocimiento en dos lenguajes, sostenido por un comentario. Es la clase de bug que este repo ya documenta que le costó una fase — y aquí el fallo sería especialmente feo: si divergen, una columna se escribiría codificada y se leería sin decodificar (o al revés), devolviendo texto JSON donde el usuario espera un objeto, sin ningún error.

⭐⭐ Y aquí es exactamente donde NACERÍA el contrato

Este tramo ya sabe lo que un contrato ODCS necesita: los nombres, los tipos de origen, la nulabilidad (IcebergIngestField.nullable) y las claves foráneas. Todo eso se resuelve, se usa para crear la tabla… y se tira: acaba como un schema en datasets, no como un contrato con semántica, calidad y clasificación.

El commit — y quién reconcilia

El plano de control envuelve al de datos, y está factorizado en una sola función (withNativeControlPlane) que comparten el POST único y el fan-out: sólo difiere el paso 2.

 ⑥′ withNativeControlPlane      lib/lakehouse/iceberg-native-write.ts
    1. abre la txn en `dataset_transactions`            → status 'open'
    2. ESCRIBE el snapshot  (el único paso que difiere entre POST único y fan-out)
    3. commitea la txn                                  → status 'committed'
    4. registra en `iceberg_sync_log` = 'succeeded'     ← lo que hace que el oráculo
                                                          de frescura diga «al día»
    5. actualiza `datasets` (row_count · schema · size · FKs · versión de esquema)

⭐⭐ Y la pieza que hace posible reconciliar: el operationId es el id de esta misma txn de ledger, y se graba DENTRO del commit Iceberg (snapshot_properties) junto a los datos. ⇒ la pregunta «¿aterrizó la MÍA?» se le contesta al catálogo, no por correlación temporal.

⭐ Los tres estados terminales, y el que ya no se escribe

Fail-loud, sin respaldo en Postgres — deliberadamente, para no crear un split-brain.

⛔⛔ Y el cierre del recorrido: el reconciliador existe y NO SE EJECUTA

lib/compute/ratify.ts tiene los verbos (ratifyTransaction, ratifyOpenTransactions), con su barrido, su antigüedad mínima, su tope de abortos por ciclo y su dryRun. Está probado (ratify-sweep.test.ts).

grep -rn "ratifyOpenTransactions" app/ lib/ scripts/ services/ | grep -v ratify
# → SÓLO su propio test. Ni cron, ni worker, ni endpoint.

Y la ventana ya mordió (medido 2026-08-12):

dataset_transactions   committed 961 · aborted 22 · ⚠️ open 5   (del 11 y 12 de agosto)
iceberg_sync_log       succeeded 223 · failed 13 · pending 1
⚠️ txn 'open' con más de 1 hora, candidatas a reconciliación: 5

⇒ Cinco escrituras en estado «no lo sé», de ayer y de hoy, con su reparador escrito, probado y sin llamar. No es una debilidad de diseño: el diseño está resuelto. Es un barrido sin disparador — la mitad más barata y la que falta.

Lo que el trazado destapa

⛔⛔ ① La ingesta NO pasa por la cara — ninguna puerta la gobierna

El escritor de producción lee LAKEHOUSE_REST_URI de su propio entorno de Railway, y apunta a Lakekeeper directamente, no a /api/iceberg.

railway variables -s warehouse-writer | grep LAKEHOUSE_REST_URI
# → https://heroic-victory-production-…   (Lakekeeper, no la cara)

⇒ Las dos puertas de gobernanza se saltan enteras: ni la de la cara («¿puede este motor?») ni la de Junction («¿puede esta persona?»). La ingesta escribe con la identidad lakekeeper-warehouse-writer contra el catálogo, y nadie le pregunta nada.

⭐ Y desde el 08-13 se sabe lo que cuesta arreglarlo, y es mucho menos de lo que parecía. Ver §«I·0 · el coste, medido».

🏁 ② El sink pg — RETIRADO (2026-08-13)

dataset-writer.ts conservaba un motor de escritura completo contra dataset_rows —PostgresBackend con INSERT multi-fila, UPSERT por (dataset_id, row_index) y COPY FROM STDIN con tabla temporal, más writeDatasetRowsPg, el control-plane que compartía con la puerta—. Esa tabla no existe desde el 2026-07-31 (migración 20261262), y su propio log seguía diciendo «dataset_rows WILL be written».

Era inalcanzable, y eso es lo que hizo barata la retirada. write-router.ts:102 hace mode = spec.maxMode sin consultar nada —la máquina de flags se fue con el plano PG— y los tres puertos del sync (sync.full · sync.incremental · sync.append) declaran iceberg_native.

⇒ Se retiró entero: −808 líneas en 6 ficheros, check:dataset-rows-seam 9 → 0 para este fichero. Y cayó con él el brazo postgres de write() en lib/compute/junction.ts, que delegaba aquí — ningún writerId que cruce la puerta puede resolver a pg (los tres que lo hacen declaran iceberg_native).

Lo que queda en su sitio es un error ruidoso, no un hueco: un sink=pg en el sync o en la puerta lanza nombrando la causa y el fichero donde se arregla. Aceptar la escritura sería prometer una durabilidad que no existe.

⚠️ Corrección de v1.1 — ingest.stream YA estaba cerrado

La v1.1 decía que ingest.stream «no está cerrado y ejecuta INSERT INTO dataset_rows contra una tabla que no existe». Es falso, y el código lo desmiente: app/api/datasets/ingest/route.ts:370-381 hace void batch y devuelve 410 ingest_stream_closed antes de tocar nada. Se cerró en D·2, con su comentario y su mensaje. El trinquete lo dice hasta en la línea del baseline («Apretado 4→3 (D·2): ingest.stream cerrado (410), su INSERT se fue») — se leyó la tabla de puertos (maxMode: 'pg') y no la ruta.

⇒ De los puertos con maxMode: 'pg': row.edit cerrado (409), ingest.stream cerrado (410), dataset.manifest mudado a Index (dataset_file_manifest, mig. 20261258). Ninguno alcanza el plano borrado.

⛔ El único resto con pulso es handleAbortTransaction (ingest/route.ts:434): un .delete() sobre dataset_rows cuyo error no se comprueba, así que falla en silencio. Cosmético, pero es lo que queda.

⚠️ ③ Spark no está en el camino

Ni Spark Connect ni el puente participan. La ingesta es Node (extracción) + PyIceberg (escritura), y el cluster no se entera. ⇒ el paradigma de cómputo del cluster cubre el SQL Editor y no la ingesta, que es donde entra el 100 % del dato nuevo.

✅ ④ Lo que sí está bien resuelto, y no hay que tocar

• El streaming es real: fetch batch → yield en la extracción y NDJSON en el transporte. Una sincronización de millones de filas no cabe en memoria en ningún punto, y eso ya está medido y en producción.
• La coordenada se acuña en Index (findOrCreateDataset), no en el motor ni en el catálogo. Es el único punto donde la ingesta toca gobernanza, y está en el sitio correcto.
• El fan-out por particiones para tablas grandes existe y es paralelo.

🏁 I·0 · EL COSTE DE ENTRAR POR LA CARA, MEDIDO (2026-08-13)

npx dotenv -e .env.local -- npx tsx scripts/warehouse/i0-ingesta-por-la-cara.ts

123 datasets · 1 inquilino CON DATO de 8 workspaces   ⚠️ los otros 7 están vacíos
warehouse que manda hoy la ingesta ...... null   ← el único con dato NO tiene espacio propio
namespaces: opacos 7 · legacy 96 · sin estampar 20
grants de `lakekeeper-warehouse-writer` . ✅ createTable · loadTable · updateTable

⭐⭐ El hallazgo: el ÁMBITO decide, y el bloqueo es UNA operación

Contra la cara en producción, con y sin prefijo de inquilino:

Las de ámbito TABLA se salvan sin prefijo y con namespace legacy, porque la cara saca el inquilino del propio ds_<hex> resolviéndolo en Index (route.ts:207-243) — una tercera vía al inquilino que ningún doc había nombrado. Las de ámbito NAMESPACE no tienen tabla que resolver, y createTable es de ésas.

(Se midió con listTables y no con createTable a propósito: es GET, no muta, y recorre exactamente la misma resolución de inquilino. Medir el nacimiento de una tabla creándola sería medirlo rompiendo la regla de la sonda.)

La reparación, y está en un sitio

ingest-client.ts mandaba tenantWarehouseNameForDataset — el espacio de almacenamiento de Lakekeeper (carbon-{env}-w-{hex}, y null para el compartido). La cara quiere el inquilino (w_{hex}), que existe siempre.

🏁 Hecho (lib/lakehouse/warehouse-token.ts, 7 tests). Y sin interruptor: el escritor ya publica a quién apunta (GET /lakehouse/config → rest_uri_host), así que el destino se deriva de él. Un INGESTA_POR_LA_CARA=true habría puesto el mismo hecho en dos sitios, y el día que discreparan ganaría el equivocado.

⭐ Lo que NO bloquea, y se creía que sí

• El namespace legacy (116 de 123 datasets). Con prefijo, la cara no mira el nombre del namespace ⇒ la migración N·1 no es prerrequisito de esto. Estaba escrito en catalog-tenant.ts:25 y ahora está medido.
• Los grants del writer: están. ⚠️ Pero sobre 1 de 8 workspaces — los otros 7 no tienen dato, así que no hay caso que medir (regla 54: una sonda que no llega al caso no lo aprueba, lo ignora).

⚠️ Dos reservas de instrumento, dichas

1. El token de la sonda es de lakekeeper-operator, no del writer. Por eso el veredicto de permisos se apoya en la decisión compuesta contra Index, que sí es autoritativa, y no en el HTTP.
2. Un 502 en la primera pasada («el emisor rechazó la credencial de la cara») no se reprodujo al reintentar. Queda como transitorio, no como hallazgo.

⛔ Y un hueco que apareció de paso

fanout-client.ts —el camino de las tablas grandes— no manda warehouse en absoluto: la ingesta particionada aterriza en el espacio que diga el entorno del escritor, no en el del inquilino. Es anterior a esto y no lo arregla I·1.

Lo que falta — con su gate

⛔⛔ I·1·b — BLOQUEADO, y no por lo que parecía (2026-08-13)

La variable está identificada y medida:

railway variables -s warehouse-writer
  LAKEHOUSE_REST_URI       = https://heroic-victory-…/catalog     ← Lakekeeper
  LAKEHOUSE_REST_WAREHOUSE = lakehouse

Pero el código de I·1·a no puede llegar a donde corre la ingesta. Medido:

Carbon Jobs (el servicio que corre los workers de polling)
  despliega desde GitHub · branch main · commit 76cb381 (2026-08-09)
origin/main == 76cb381                     ⇐ EXACTAMENTE lo que corre
esta rama .. 142 commits POR DELANTE de main

⇒ El derivador vive en una rama que el worker no ve. Apuntar la variable hoy pondría la cara delante de un worker que sigue mandando el token de Lakekeeper ⇒ 403 en createTable: el fallo medio-vivo, provocado a propósito.

⭐ Y el reparto de despliegues no es uniforme, que es lo que lo hace fácil de no ver: Vercel se despliega desde el árbol local (por eso hay que hacerlo desde un worktree limpio), pero Carbon Jobs se despliega desde main. La misma línea de código llega a un sitio y no al otro.

Las tres salidas

⇒ Decisión del owner. Lo que NO se puede es apuntar la variable y ver qué pasa: el modo de fallo no aparece hasta que alguien conecta una fuente nueva.

⚠️ Y un cabo suelto para cuando se apunte: LAKEHOUSE_REST_WAREHOUSE=lakehouse seguiría siendo el respaldo del escritor para cualquier camino que no mande warehouse. La cara no lee lakehouse como inquilino ⇒ sin prefijo ⇒ 403 al crear. Hoy es inerte (ingest-client siempre manda uno y el fan-out está apagado), pero es exactamente la forma de mina que este doc acaba de desarmar en otro sitio.

⭐ I·2 — contestado EN CÓDIGO: el «nunca» está caducado (2026-08-13)

ratify.ts:126 dice, sobre las escrituras upsert:

Es cierto mientras firme pyiceberg. writer.py:556 hace table.upsert(...), que no acepta snapshot_properties. Pero por la cara firma la cara: updateTable lleva un add-snapshot en updates[] —cualquier commit de datos lo lleva—, que es justo lo que añadeSnapshot busca, y el operation_id sale de operacionEnVuelo, que encuentra la txn de ledger porque withNativeControlPlane la abre antes del data-plane.

⇒ Es la cuarta vez que un guardián con condición de retirada escrita caduca (regla 49).

⭐⭐ Y la mitad que nadie había nombrado

Que la cara firme no basta. ratify no llega a preguntar al catálogo: se para antes, en isAttributable(mode) (attribution-client.ts:116), que devuelve false para upsert por definición estática:

export function isAttributable(writeMode): boolean {
  return writeMode !== 'upsert';   // ⛔ una AFIRMACIÓN sobre pyiceberg, no una medición
}

⇒ I·2 son dos piezas, no una: que la cara firme (mecanismo, ya razonado) y retirar esa afirmación cuando se mida (guardián). Retirar sólo la segunda declararía verificable lo que no lo es; dejar sólo la primera seguiría diciendo unverifiable sobre commits que ya llevan su marca.

Gate de I·2: una escritura upsert por la cara cuyo snapshot lleve carbon.operation-id = el id de su txn de ledger, y ratify diga committed. ⛔ Exige I·1·b.

Historial