El replace destructivo — recuperar el linaje de los datasets nativos

Estado (2026-07-29)

El canary es la env-var LAKEHOUSE_REPLACE_PRESERVES_HISTORY del servicio ml-runner. A 0 (por defecto) el comportamiento es el destructivo de siempre, byte por byte; a 1 se conserva el linaje. GET /lakehouse/config lo reporta, así que el modo real se pregunta, no se supone.

1 · La evidencia

En el código (services/ml-runner/app/lakehouse/writer.py), dos sitios:

El propio docstring lo declara: «replace → drop any existing table and CREATE FRESH … the add_files commit is then the fresh table's single snapshot».

Medido en vivo (scripts/dataspaces/jop-e2e-loop.ts, sobre animal_events):

• tras un replace, compensate responde no-prior-state — no hay snapshot padre;
• una segunda escritura hace que la primera pase de landed:true a landed:false: su snapshot fue dropeado y ratify ya no puede confirmarla;
• la tabla vuelve siempre a snapshots=1.

2 · Por qué está así, y por qué el motivo no se sostiene

El drop es la vía de escape ante un cambio de esquema: «a schema change must re-snapshot» (writer.py:378).

Ese razonamiento es falso en Iceberg. La evolución de esquema es metadata-only —añadir, borrar, renombrar, reordenar y ensanchar tipos— resuelta en milisegundos sobre IDs de columna estables, sin reescribir un solo fichero de datos, independientemente del tamaño de la tabla. Es la característica insignia del formato.

Y no es cuestión de estilo: la documentación presenta el reemplazo atómico como algo que conserva la historia («creates a new snapshot … but keeps table history»), y existe un issue en el propio proyecto titulado «Replace table shouldn't clear table history» (apache/iceberg#2233). La comunidad lo trata como defecto, no como alternativa.

3 · Qué se pierde hoy (más de lo que parece)

1. Time-travel — inexistente para todo dataset nativo. Se está pagando Iceberg y tirando su principal ventaja en cada full sync.
2. Auditoría — no se puede responder "¿cómo estaba esta tabla ayer?".
3. Compensación — compensate no puede fijar un estado previo que ya no existe.
4. Ratificación retroactiva — ratify deja de poder confirmar cualquier operación anterior al último replace.
5. Y es el mecanismo del riesgo ya conocido: que un sync.full posterior «oblitere en silencio» un write de DuckDB no es una metáfora — es literalmente este drop_table.

4 · La forma correcta: DOS cosas, no una

Sustituir el drop por overwrite a secas no basta — no cubre el caso que motivó el drop. La semántica de CREATE OR REPLACE TABLE son dos operaciones:

Ambas existen en pyiceberg.

El coste real, y cómo se paga

Conservar historia significa que los ficheros antiguos permanecen hasta expirar — real, y más con un full sync frecuente sobre tablas grandes. La respuesta de Iceberg no es destruir de forma anticipada, es una política de retención: expire_snapshots como job de mantenimiento, con history.expire.max-snapshot-age-ms y min-snapshots-to-keep.

Es un dial, no un binario. Incluso conservando unos pocos snapshots se recupera compensación, ratify y auditoría en la ventana reciente, que es donde importan.

El caso residual donde el drop SÍ es legítimo

union_by_name cubre cambios aditivos y compatibles; Iceberg solo permite ensanchar tipos. Un cambio genuinamente incompatible —estrechar un tipo, o reusar un nombre con otro significado— no es expresable como evolución metadata-only.

El manejo correcto no es dropear siempre, sino evolucionar cuando se puede y dropear solo cuando el cambio es de verdad incompatible, como decisión explícita y registrada. El writer ya tiene el gancho: _schema_matches (:284) sabe cuándo difiere — simplemente elige siempre la rama destructiva.

5 · GATE R·0 — la incógnita, cerrada

Nuestro path real no usa overwrite: usa add_files sobre chunks Parquet (init → stage → commit). Para preservar historia hace falta borrar todo + add_files en una sola transacción. Si eso no fuera atómico, cambiaríamos un problema por otro peor.

Verificado leyendo el fuente de pyiceberg 0.11.1 (la versión que writer.py ya documenta como probada; requirements.txt pinea un RANGO >=0.8,<1.0, así que ahora GET /lakehouse/config reporta la versión real del build en vez de dejarla a la suposición):

1. Transaction.delete() y Transaction.add_files() acumulan sus updates en el mismo _updates; commit_transaction() emite UN solo _do_commit → una sola llamada commit_table al catálogo → un solo CAS sobre metadata.json. La atomicidad se conserva.
2. Cada uno produce su propio snapshot (DELETE y APPEND): una operación pasa a ser un tramo de dos, no un punto. El encadenado es correcto porque Transaction.table_metadata se recalcula sobre los updates acumulados, así que el add_files tiene como padre al delete.
3. La prueba más fuerte es que ya corre en producción: pyiceberg implementa Transaction.overwrite exactamente así —self.delete(filtro) seguido de un append en la misma transacción— y write_snapshot usa table.overwrite() contra este mismo Lakekeeper hoy. El mecanismo no es nuevo; lo nuevo es de dónde salen los ficheros.
4. Sobre una tabla recién creada el delete no afecta a ningún fichero y no produce snapshot (_DeleteFiles._commit devuelve vacío), así que el primer write de un dataset sigue dejando uno solo.

La otra mitad — que Lakekeeper acepte un commit_table con dos add-snapshot— se midió (R·4, 2026-07-29) y también pasa. Con el canary a 1 sobre animal_events de main.test, 18/18 verde:

Ese +2 por operación es, literalmente, la factura de R·3: sin expiración, cada full sync suma dos snapshots para siempre.

5.1 · Lo que R·4 destapó: el orden del array de snapshots no es fiable

El harness dio 5 fallos y no eran ruido. Medido con ocho llamadas idénticas a /lakehouse/operation-status: el catálogo devolvió metadata.snapshots en dos órdenes distintos, así que «el último del array» tanto era el alta (total-records=4) como el vaciado (total-records=0).

La fragilidad no la introduce el linaje. El find_operation_snapshot anterior ya recorría el array en orden inverso; solo era invisible mientras cada operación fuera un único snapshot. El tramo de dos la volvió portante.

Arreglado ordenando por sequence_number, que es la autoridad: Iceberg v2 lo obliga a ser estrictamente monótono (rechaza un snapshot con secuencia menor que la última) y pyiceberg lo avanza con cada snapshot dentro de la misma transacción, así que ordena bien incluso los dos de un mismo commit. timestamp_ms y snapshot_id solo desempatan en tablas v1. Nada vuelve a depender de la posición.

De paso, operation-status devolvía snapshot_count solo cuando no encontraba la operación: había que preguntar por una operación inexistente para saber si la tabla acumulaba historia. Ahora se devuelve siempre, junto a operation_snapshot_count (1 el camino normal, 2 el reemplazo con linaje).

El efecto colateral que habría roto compensate en silencio

Al pasar una operación de un snapshot a dos, compensate seguía cogiendo parent_snapshot_id del snapshot publicado — que ahora es el vaciado intermedio, la tabla vacía. Habría fijado un tag sobre la nada y parecido que funcionaba. Corregido: find_operation_snapshots() devuelve el tramo completo, ratify lee el último (lo publicado) y compensate el primero (para llegar al padre = el estado real anterior).

Dos cosas más que arregla, y que no estaban en el plan

• Una ventana de pérdida de datos. El fan-out hacía init (drop+create) → stage → commit en peticiones distintas. Si el stage o el commit fallaban, la tabla ya estaba dropeada y el dataset quedaba vacío. Ahora la tabla anterior sobrevive intacta hasta el commit atómico.
• El path fan-out era INATRIBUIBLE. writeIcebergNativeFanout recibía el operationId del control-plane y lo descartaba, y /lakehouse/commit ni siquiera aceptaba el campo: ninguno de sus commits llevaba marca de procedencia, así que ratify no podía confirmar ni uno. Cableado de punta a punta.

6 · Plan y qué se hizo

• R·0 · GATE — verificar la atomicidad. ✅ Cerrado en el eje pyiceberg (§5). La mitad de Lakekeeper solo se mide en vivo.
• R·1 · Evolución de esquema en vez de drop. ✅ open_or_create_native_table: si el esquema difiere y el cambio es aditivo (_schema_evolvable: las columnas existentes son un prefijo de las nuevas), se aplica update_schema().union_by_name() —metadata-only— y la tabla se conserva. Solo un cambio realmente incompatible cae a la rama destructiva, y entonces se registra como incidente (logger.error con el delta de columnas), no como un warn que nadie lee.
• R·2 · Reemplazo preservando linaje. ✅ commit_add_files(replace=True) publica delete-all + add_files en una Transaction. El vaciado viaja en el commit, no en el init — de ahí que se cierre la ventana de pérdida de datos. Se marcan los dos snapshots con el operation_id. check_duplicate_files=False: conservar la tabla haría que ese chequeo leyera todos sus manifests antes de cada full sync, y es innecesario porque cada chunk se nombra con uuid4 bajo el data/ de la propia tabla.
• R·3 · Política de retención. ✅ Implementada — con una corrección de fondo al plan (§7.1). La política se declara en cada tabla con las propiedades estándar de Iceberg (así cualquier motor que recolecte honra la misma regla) y la ejecuta expire_native_snapshots, porque pyiceberg define esas propiedades pero no las aplica. Defaults: 7 días Y mínimo 5 snapshots, aplicando el más conservador — el mínimo existe justo para nuestro caso, ya que con un full sync frecuente todo es «viejo» enseguida y solo por edad se vaciaría la ventana de recuperación entera. Piezas: POST /lakehouse/expire-snapshots (dry_run por defecto) · lib/lakehouse/retention-client.ts · lib/workers/lakehouse/retention-scheduler.ts (opt-in ENABLE_LAKEHOUSE_RETENTION, diario, y arranca en dry-run salvo LAKEHOUSE_RETENTION_APPLY=true) · scripts/dataspaces/expire-snapshots.ts.
  • Lo fijado por un tag no se expira nunca. pyiceberg protege las cabezas de rama y los snapshots con tag, que es exactamente lo que compensate fija: su promesa de que el estado compensado «sobrevive a la expiración» se sostiene porque el motor lo impide, no por convención.
• R·4 · Verificación. ✅ jop-e2e-loop.ts pregunta al runner en qué modo está (GET /lakehouse/config → replace_preserves_history) y afirma lo que corresponda: en modo destructivo fija la realidad vieja, en modo linaje afirma las aserciones invertidas (pinned con parentSnapshotId = el snapshot anterior; la 2ª escritura da superseded y la 1ª operación sigue con landed:true). Un harness que solo supiera una de las dos cosas mentiría la mitad del tiempo. Corrido en las dos direcciones: 17/17 con el canary apagado (⇒ el despliegue es neutro) y 18/18 encendido, tras destapar y arreglar el bug de orden de §5.1.

Cobertura en frío

services/ml-runner/tests/test_lakehouse_replace_lineage.py (20 tests verdes con los de ingest) fija con dobles la forma del commit —una sola transacción, delete antes que add_files, marca en ambos, check_duplicate_files=False—, las cuatro decisiones de open_or_create_native_table (conserva / evoluciona / dropea solo si es incompatible / canary apagado = idéntico a antes) y que el orden del tramo no dependa del array (le pasa los snapshots al revés y exige el orden correcto). No necesita pyiceberg: la prueba semántica es R·4.

⚠️ Al tocar estos tests salió que la suite de ml-runner llevaba roja en main: el hilo del operation_id (JOP) añadió un argumento a commit_add_files y 5 tests se quedaron con la firma vieja. No lo detectó nadie porque no hay workflows de CI en el repo — los tests de services/ solo corren a mano. Arreglado; el hueco de CI, no.

7 · Decisiones para el owner

Resueltas por el camino:

• (3) Alcance → canary, como se sugería: una env-var del ml-runner, apagada por defecto, con R·4 como gate.

• (2) Cambio de esquema incompatible → no se falla loud todavía. Se eligió el escalón intermedio: evolucionar cuando se puede y, cuando no, dropear registrándolo como incidente. Fallar en seco convertiría un sync que hoy se auto-repara en una intervención manual, y eso merece decidirse con datos: el log del incidente dirá si ocurre alguna vez. Si aparece, el cambio a raise es de una línea.

• (1) Retención por defecto → 7 días Y mínimo 5 snapshots, como se sugería, declarado en la tabla y configurable por entorno (LAKEHOUSE_SNAPSHOT_MAX_AGE_MS / LAKEHOUSE_SNAPSHOT_MIN_TO_KEEP).

• (2) Quién lo ejecuta → un scheduler diario nuevo en el servicio de workers, opt-in y en dry-run por defecto. Ya no hay que confiar en que «quien compacte respete las propiedades»: nadie compacta, así que el ejecutor es nuestro.

7.1 · La corrección de fondo: expirar NO libera espacio

Verificado en el fuente de pyiceberg 0.11.1 al implementar R·3, y es lo contrario de lo que este documento asumía en §4:

• ExpireSnapshots existe y protege refs (tags y cabezas de rama). ✅
• Pero RemoveSnapshotsUpdate es metadata-only: quita los snapshots de la lista, del snapshot_log, de los refs y de las estadísticas — y no borra ni un fichero de datos ni un manifest. No existe remove_orphan_files en esta versión.
• Las propiedades history.expire.* están definidas como constantes pero nada las lee: solo las muestra la CLI.

Lo que R·3 sí consigue: acotar la ventana recuperable (que es lo que hace que las garantías de compensate sean finitas y no una promesa vaga) y el tamaño de metadata.json. Lo que NO consigue: bajar la factura de R2.

Decisión abierta que esto deja sobre la mesa: quién recoge los ficheros huérfanos. Tres caminos, y ninguno es gratis:

1. GC del lado de Lakekeeper — hay que verificar si lo hace al recibir un remove-snapshots; no está medido (se mide comparando el prefijo de la tabla en R2 antes y después de un --apply).
2. Un barredor propio que liste el data/ de la tabla y borre lo que ningún manifest referencia. Es el clásico orphan file cleanup, y es peligroso de hacer a mano: borrar un fichero aún referenciado corrompe la tabla.
3. Lifecycle rules de R2 — no sirve tal cual: no distingue un fichero referenciado de uno huérfano.

(Nota de método: la sonda para (1) no se pudo correr — las credenciales S3 de .env.local traen un marcador pegado al valor y, ya limpias, la firma no valida. Hacen falta credenciales de lectura de R2 vigentes.)

7.2 · R·3 ejecutado en vivo (2026-07-29)

scripts/dataspaces/expire-snapshots.ts --apply contra el runner desplegado:

• 5 namespaces barridos (datasets, bench, main, main.test, main.default) y 133 tablas. ⚠️ Antes de arreglar el barrido esto habría cubierto uno solo — ver §7.3.
• 3 snapshots expirados bajo la política por defecto (7 días, mínimo 5): una tabla 6 → 5, tres veces. Lakekeeper aceptó el RemoveSnapshotsUpdate.
• Tablas íntegras después: table_exists=true, snapshot_count=5, cabeza viva.
• Idempotente: el segundo pase encuentra 0.

Hallazgo colateral: las tres tablas afectadas no tienen fila en datasets — son tablas huérfanas en el catálogo, y por eso eran lo más rancio del warehouse. Es decir, el primer efecto real del job de retención cayó sobre lo que nadie está mirando. Cruza con el paradigma de retirada de huérfanos (memoria orphaned-retirement-dedup): mientras existan, son a la vez los mayores consumidores de historia y los que menos falta hace conservar.

7.3 · El namespace, por tercera vez

Corriendo R·3 en vivo el barrido reportó namespace: datasets sobre una tabla que vive en main.test, y devolvió «0 tablas». Eso no da error: se lee como nada que expirar cuando lo que pasó es miré donde no era.

Es el mismo fallo por tercera vez —F3.2 en la cintura, el cliente de atribución, y ahora el de retención—, así que la resolución dejó de reimplementarse por cliente: vive en lib/lakehouse/dataset-namespace.ts y el cliente de atribución pasó a usarlo en vez de su copia privada. Y el barrido completo, que cubría un namespace, ahora los recorre todos (list_namespaces recursivo).

Entregable. Cruza con: junction.md (JOP) · junction-b1b2-control-plane-repair.md · scripts/dataspaces/jop-e2e-loop.ts (la medición). Fuentes: Spark DDL · apache/iceberg#2233 · PyIceberg API · Schema evolution.