Fase 4 — Aterrizar el dato consumido en el lakehouse (Iceberg gobernado)

1. El problema central (y el insight)

El transfer deja un Parquet suelto en R2 (bucket destino). Para que sea un dataset de primera clase de Carbon hay que: crear el artifact (datasets + project_files), crear la tabla Iceberg, y registrar los datos con el control plane (dataset_transactions ledger + iceberg_sync_log + stats).

Insight de la auditoría (clave para el alcance): existe add_files (registrar un Parquet ya en R2 sin reescribir — commitLakehouseFiles → /lakehouse/commit). Es tentador por zero-copy. PERO el Parquet del partner no trae las columnas de identidad de Carbon (__row_index, __row_id) que las escrituras nativas sí ponen y de las que depende iceberg_sync_log.has_identity=true (keyset pagination, read-router). Un add_files directo de un Parquet ajeno produciría un dataset degradado (sin identidad) y marcaría has_identity incorrectamente.

→ La decisión de alcance gira sobre identidad vs zero-copy.

2. Decisión de encaje — dos estrategias

Recomendación: B. El valor de Carbon es tener datasets gobernados y consistentes (lineaje, paginación, read-router). Un dataset "de segunda" (sin identidad) rompería esa promesa.

3. Sub-problema: ¿dónde/con qué clave aterriza el Parquet?

Hallazgo Fase 2: el sink S3 de EDC ignoró el prefijo del keyName (dataspace-inbox/sample.parquet → aterrizó sample.parquet en la raíz del bucket). Para saber qué objeto produjo cada transfer:

• En consume: fijar el dataDestination a un bucket de inbox de dataspaces (en el R2 del lakehouse) con un keyName plano y único por transfer (<transferId>.parquet, sin "/") → esquiva el bug del prefijo.
• En el worker (al COMPLETED): verificar por listing el objeto real (robusto si el sink cambia la clave), y pasar esa ruta R2 a ingest-parquet.

(Comprobar empíricamente el comportamiento exacto del sink con keyName plano — igual que hicimos con v4 — es parte de la iteración.)

4. Alcance de la iteración

Dentro:

• Endpoint ml-runner /lakehouse/ingest-parquet: dado {dataset_id, namespace, source_path (R2), schema?} → lee el Parquet, infiere/valida schema, reescribe snapshot nativo (identidad + chunked add_files internos), devuelve {identifier, rows, snapshot_id, size_bytes}.
• lib/dataspaces/lakehouse-landing.ts — landConsumedTransfer(row): inspecciona schema → createDatasetArtifact → initLakehouseTable → withNativeControlPlane(callback = ingestParquet).
• Hook en el worker (sync-worker.ts, paso COMPLETED): COMPLETED → LANDING → LANDED (best-effort), guarda target_dataset_id.
• consume: fijar dataDestination al inbox del lakehouse con keyName único.

Fuera (a propósito):

• ❌ Zero-copy real / DataSource Lakekeeper (compartir tabla sin mover) → Fase 5.
• ❌ Append/incremental/upsert del mismo asset en el tiempo → primer corte = replace one-shot (cada transfer = un snapshot).
• ❌ UI de la pestaña Dataspaces → Fase 7.
• ❌ Mapear el dataset consumido a la ontología/objetos → fases de ontología.

5. Piezas y reúso

A crear (poco, aditivo):

Reutilizado TAL CUAL (cero cambios al core):

6. Contratos concretos

/lakehouse/ingest-parquet (nuevo):

POST  { dataset_id, namespace, source_path: "s3://…/<transferId>.parquet", schema?: BASEField[] }
  →   { identifier, rows, snapshot_id, location, size_bytes }

(Si schema viene, se valida/usa; si no, se infiere del Parquet — pyarrow → BASE.)

landConsumedTransfer(row: DataspaceTransfer) (nuevo, Node):

1. sourcePath ← descubrir el objeto real del transfer (row.config.dest + listing)
2. schema     ← inspeccionar (endpoint o dentro de ingest-parquet)
3. { datasetId } ← createDatasetArtifact(workerDbClient, { workspaceId: row.workspace_id,
                      projectId, name: `EDC: ${row.asset_id}`, service: 'edc_consumer',
                      schema, metadata: { edc_transfer_id: row.edc_transfer_id, counterparty: row.counterparty }})
4. initLakehouseTable(datasetId, resolveIcebergSchema(schema), { writeMode: 'replace' })
5. withNativeControlPlane(
     { datasetId, workspaceId: row.workspace_id, columns: schema, writeMode: 'replace', namespace,
       metadata: { source: 'edc_consumer', edc_transfer_id: row.edc_transfer_id } },
     async () => ingestParquet(datasetId, namespace, sourcePath),  // → NativeIngestOutcome
   )
6. UPDATE dataspace_transfers SET target_dataset_id = datasetId, state = 'LANDED'

Estados (columna state, sin migración nueva): COMPLETED → LANDING → LANDED (o COMPLETED + error si el landing falla — el transfer siguió OK; el landing se reintenta). target_dataset_id ya existe.

Proyecto destino: para el primer corte, un proyecto/carpeta "Dataspaces" por workspace (o el default del workspace). Resolver projectId es un detalle menor (reutilizar el resolver de proyecto por defecto).

7. Runbook

1. Empírico: confirmar el comportamiento del sink con keyName plano único (¿respeta <transferId>.parquet?). Ajustar la estrategia de descubrimiento.
2. ml-runner: /lakehouse/ingest-parquet (leer R2 Parquet → snapshot nativo). Probar aislado con un Parquet en R2.
3. Node: lib/dataspaces/lakehouse-landing.ts + cliente del endpoint.
4. Wire: hook en el worker (COMPLETED → landConsumedTransfer → LANDED) + consume fija el dataDestination.
5. E2E: consume → worker → transfer COMPLETED → aparece un dataset en el workspace con las filas del Parquet, gobernado (ledger + sync_log + stats). Verificar con un smoke (patrón de los de Fase 3) contra el conector + Supabase real.

8. Definition of Done

• /lakehouse/ingest-parquet registra un Parquet de R2 como snapshot Iceberg nativo (con identidad).
• landConsumedTransfer crea el dataset (datasets+project_files) y lo llena vía withNativeControlPlane.
• El worker, en COMPLETED, aterriza y setea target_dataset_id + state=LANDED (best-effort; error no pierde el transfer).
• E2E: un consume acaba con un dataset visible y legible en el workspace (read-router lo lee como cualquier dataset).
• Cero cambios al control plane / ledger / oracle; solo reúso.

9. Riesgos

10. Qué desbloquea

• El dato consumido por el dataspace es ya un dataset gobernado de Carbon → pipelines, ML, ontología, lineaje inter-org (el transfer queda ligado al dataset por edc_transfer_id/target_dataset_id).
• Fase 5 (publicar / zero-copy) reutiliza el mismo esqueleto de landing/registro en sentido inverso.
• Fase 7 (UI): la pestaña Dataspaces ya puede mostrar "consumido → dataset X".

11. Referencias de la auditoría

Siguiente acción al ejecutar: Paso 1 del runbook — confirmar empíricamente la clave de aterrizaje del sink, luego el endpoint /lakehouse/ingest-parquet.