Pipeline: cross-workspace-csv-merge

• Scopo: demo di flusso cross-workspace: legge test-etl.csv da due workspace Noeva, ne produce la union deduplicata delle righe, e la riscrive come nuova versione su entrambi i workspace.
• Owner: —
• Cadenza: on-demand (demo).

Flusso dati

0/4

Scegli una scena e premi ▶ per guardarla scorrere, oppure → per andare avanti un passo alla volta.

Lo scopo dei nodi

Nodo	Tipo	Scopo
read-workspace-a	source	Legge il file condiviso dal primo workspace. Se lì il file non esiste ancora, il workspace non porta righe e l'unione parte lo stesso.
read-workspace-b	source	Legge lo stesso file dal secondo workspace. Se lì il file non esiste ancora, il workspace non porta righe e l'unione parte lo stesso.
union-workspaces	union	Mette in fila le righe dei due workspace, prima tutte quelle del primo e poi quelle del secondo. L'ordine fisso fa sì che, a dati fermi, ogni run produca lo stesso identico file.
dedup-rows	transform	Toglie le righe ripetute, tenendo la prima copia di ognuna. Il risultato torna in tutti e due i workspace: senza questo passo il file raddoppierebbe a ogni run.
encode-csv	transform	Ricompone le righe in un unico file, con tutte le colonne presenti in almeno uno dei due workspace. Una colonna che un workspace non ha resta vuota nelle sue righe, così nessun dato va perso.
workspace-a	sink	Salva il file unito come nuova versione nel primo workspace. Se il contenuto è uguale all'ultima versione, non ne crea un'altra.
workspace-b	sink	Salva il file unito come nuova versione nel secondo workspace. Se il contenuto è uguale all'ultima versione, non ne crea un'altra.

Il grafo qui sopra è disegnato dalla topologia reale della pipeline (getSpec()): nodi, archi, dead-letter e ramo catch sono quelli eseguiti. Premi ▶ per far passare un record, o → per avanzare un passo alla volta.

Una source per workspace (read-workspace-a, read-workspace-b) legge il file e ne emette le righe così come sono; un union le accoda, A prima di B; un dedup su riga intera tiene la prima occorrenza di ogni riga; un nodo di encode serializza in CSV con l'unione delle colonne come intestazione; il byte stream fa fan-out su due sink Noeva, uno per workspace.

Sorgenti

• test-etl.csv su workspace A (il primo servito dall'istanza) e workspace B (NOEVA_MERGE_SECONDARY_WORKSPACE_ID, default 8793575e-7546-489e-91c9-3004918361fb), via le API files di noeva-server-api (src/connectors/noeva).
• Lette da workspaceCsvSource (src/pipelines/cross-workspace-csv-merge/workspace-source.ts), in streaming. La memoria cresce con le righe distinte: il dedup ne tiene le chiavi, e encode-csv le trattiene per conoscere l'unione delle colonne prima di scrivere. Il tetto è maxKeys: 100 000 sul dedup: oltre 100 000 righe distinte la run si ferma con un errore che nomina l'opzione, invece di scrivere un merge a metà sui due workspace.
• Tollerante all'assenza: se il file non esiste in un workspace, quel workspace contribuisce 0 righe (warning nel log), così il merge parte anche al primo giro. Se manca in entrambi, l'encode non produce byte e i sink caricano un file vuoto (0 byte); alla run successiva è identico e l'upload si salta.

Sink

• test-etl.csv su entrambi i workspace, in modalità new-version (noevaFileSink). Il sink è idempotente: salta l'upload se il contenuto è identico all'ultima versione.

Trasformazioni e regole di business

• Merge = union deduplicata su riga intera, in due nodi:
  • union-workspaces — nodo union in modalità concat: prima tutte le righe di A, poi quelle di B;
  • dedup-rows — transform dedup con keep: "first" e chiave rowKey (src/pipelines/cross-workspace-csv-merge/row-key.ts, serializzazione delle colonne non vuote ordinate per nome): tiene la prima occorrenza di ogni riga distinta. Una colonna assente e una vuota sono la stessa riga, come lo sono nel CSV scritto.
• Encode = unione delle intestazioni (encode-csv, src/pipelines/cross-workspace-csv-merge/encode-csv.ts): l'intestazione è l'unione delle colonne di tutte le righe, nell'ordine in cui compaiono la prima volta (quelle di A, poi quelle che solo B ha). Una riga senza una colonna la scrive vuota. Senza righe non scrive nulla, nemmeno l'intestazione.
• L'ordine fa parte del contratto, ed è per questo che il union è in concat e non in interleave: i sink new-version saltano l'upload solo se i byte sono identici all'ultima versione, e un ordine che cambiasse fra una run e l'altra produrrebbe una versione nuova a ogni run anche a dati fermi.
• Idempotenza (importante): scrivere merge(A,B) dentro sia A sia B è un feedback loop lettura/scrittura. Dedup e ordine fisso garantiscono un punto fisso: dopo la prima run A e B diventano identici; alla run successiva dedup(A ∪ B) è già uguale, byte per byte, all'ultima versione → il sink new-version salta l'upload. Un semplice concat, invece, raddoppierebbe i dati a ogni run, e un sink che duplica i dati a ogni riesecuzione non è ammesso.

Intestazioni diverse fra A e B

L'output sovrascrive le sorgenti, quindi una colonna persa in scrittura è un dato perso: invece di fermarsi quando le intestazioni differiscono, la pipeline ne scrive l'unione. Così un file che guadagna una colonna in un workspace la propaga all'altro, e il merge raggiunge il punto fisso alla prima run anche in quel caso.

Caso	Comportamento
Colonna presente solo in B	Resta nell'output, vuota nelle righe di A
Stesse colonne in ordine diverso	Ordine di prima comparsa: quello di A, poi le colonne nuove di B
{id,val} in A, {id,val,extra: ""} in B	Una riga sola: la colonna assente vale come vuota
Riga corta (meno campi dell'intestazione)	Ammessa: la source mette a vuoto le colonne mancanti
Riga lunga (più campi dell'intestazione)	ExtractError fail-fast, nessuna scrittura: il campo in più non ha una colonna
Nome di colonna duplicato o __proto__	ExtractError fail-fast, nessuna scrittura: se ne terrebbe un valore o nessuno
Colonna che nessuna riga sopravvissuta porta	Esce dall'intestazione: succede solo se è vuota ovunque, nessun valore si perde

Una colonna esce dall'intestazione solo quando il suo file non ha righe, o quando ogni sua riga coincide con una già vista (e quindi quella colonna è vuota in tutte).

Limite noto, indipendente dalle intestazioni: in un file a una sola colonna, una cella vuota si scrive come riga vuota, e il decode la salta alla run dopo.

Contratti (schema Zod)

• Input/Output: csvRowSchema = z.record(z.string(), z.string()) (src/pipelines/cross-workspace-csv-merge/schema.ts) — riga CSV schema-agnostica (mappa colonna→valore). Un decode che producesse qualcosa di diverso è un errore fail-fast (ExtractError), nessuno scarto silenzioso.

Colonne nel piano (getSpec): essendo la riga schema-agnostica (z.record), le colonne non sono introspezionabili e vengono dichiarate esplicitamente sugli stub delle due letture (CSV_COLUMNS = [id, val], l'header noto di test-etl.csv). Da lì si propagano a union-workspaces, dedup-rows, encode-csv e ai due sink (workspace-a, workspace-b) nella UI del DAG. Sono rappresentative: con intestazioni diverse fra A e B il file scritto ne ha di più (vedi sopra).

Descriptor nel piano: gli stub di getSpec usano le stesse funzioni dei connettori veri (workspaceCsvDescriptor, noevaFileSinkDescriptor), quindi il DAG mostra per ogni nodo gli stessi dettagli della run. Un test lo verifica.

Ogni riga porta la variabile da cui arriva e la sua origine: base URL e workspace A dal client Noeva (NOEVA_*), workspace B e file dalle variabili della pipeline (NOEVA_MERGE_*). Il workspace B di default compare con origine default. La riga credenziale dice solo se NOEVA_API_KEY è impostata. Senza .env il client Noeva è d'esempio, con le sue variabili mancanti; le variabili NOEVA_MERGE_* stanno anche nella configurazione della pipeline del piano.

Per pnpm run etl env check la pipeline dichiara le sole NOEVA_MERGE_* (environment): le NOEVA_* del client sono del core, che le dichiara da sé.

Parametri ed env

• ctx.params: nessuno.
• Variabili d'ambiente (vedi .env.example):
  • NOEVA_WORKSPACE_IDS — richiesto: deve elencare sia il workspace A sia il workspace B (l'NOEVA_API_KEY è una sola e vale per entrambi).
  • NOEVA_WORKSPACE_ID — workspace A quando NOEVA_WORKSPACE_IDS non è impostata; con la lista, il workspace A è il suo primo id e questa variabile non conta.
  • NOEVA_API_BASE_URL — endpoint dell'API, condiviso dai due workspace; nel piano (getSpec) compare nel descriptor dei nodi.
  • NOEVA_MERGE_SECONDARY_WORKSPACE_ID — workspace B (default 8793575e-…).
  • NOEVA_MERGE_FILE_NAME — nome file (default test-etl.csv). Le variabili NOEVA_MERGE_* sono dichiarate in MANIFEST_CROSS_WORKSPACE (config.ts), quelle NOEVA_* in MANIFEST_NOEVA: una run con variabili mancanti si ferma con un ConfigError che le elenca tutte.
• Validazione workspace: ogni workspace target deve essere tra quelli configurati. Puntare a un workspace non presente in NOEVA_WORKSPACE_IDS è un errore fail-fast (ConfigError), per non avviare processi su workspace non previsti.

Gestione errori

• File assente in un workspace → tollerato (0 righe + warning).
• CSV malformato o errore di lettura → ExtractError (fail-fast), che nomina il workspace: nel report il nodo fallito è la source di quel workspace. I sink Noeva lasciano passare invariato un errore arrivato da monte, e raccolgono tutti i byte prima di caricare: se una lettura fallisce, nessun workspace riceve una versione nuova.
• Workspace non configurato → ConfigError prima di qualsiasi I/O.
• Nessun dead-letter: la demo è fail-fast (i record CSV sono schema-agnostici e sempre validi per costruzione). I duplicati scartati dal dedup non sono errori: il dedup usa duplicates: "drop", e si contano come differenza sul nodo.
• Riga con più campi dell'intestazione, o nome di colonna duplicato → ExtractError (fail-fast), come un CSV malformato: il codec della source gira con unmappedFields: "error" invece di perdere quei valori, che riscrivere il file cancellerebbe. Vedi Intestazioni diverse fra A e B.

Report

Ogni workspace è un nodo, quindi il report della run dice quante righe ha dato ciascuno: extracted di read-workspace-a e di read-workspace-b. Il totale extracted della run è la loro somma, prima della dedup; i duplicati scartati sono extracted − written di dedup-rows.

Test

• src/pipelines/cross-workspace-csv-merge/characterization.test.ts — l'output consegnato ai due workspace, passando da run(): byte esatti (righe e ordine), punto fisso, file assente, fail-fast su CSV malformato; intestazioni diverse fra A e B (colonna solo in B, colonna assente ≡ vuota, punto fisso, righe corte, riga lunga, colonna duplicata).
• src/pipelines/cross-workspace-csv-merge/config.test.ts — default e override delle variabili NOEVA_MERGE_*, con la loro origine.
• src/pipelines/cross-workspace-csv-merge/encode-csv.test.ts — unione delle colonne, byte invariati a intestazione uguale, nessun output senza righe.
• src/pipelines/cross-workspace-csv-merge/workspace-source.test.ts — lettura di un workspace, tolleranza all'assenza, fail-fast su CSV malformato, riga lunga e colonna duplicata, righe corte completate con celle vuote, descriptor.
• src/pipelines/cross-workspace-csv-merge/row-key.test.ts — chiave di dedup, colonna assente ≡ vuota.
• src/pipelines/cross-workspace-csv-merge/pipeline.test.ts — e2e (stesso CSV su entrambi i sink, ordine, dedup, idempotenza, nessuna riga), conteggi per workspace nel report, errore attribuito al nodo del workspace coi sink Noeva veri, topologia getSpec e descriptor del piano uguali a quelli dei connettori veri; variabile e origine delle righe, workspace B di default con origine default, client Noeva d'esempio senza .env.
• src/connectors/noeva/workspace-registry.test.ts — validazione workspace configurati.

Versioni

La storia della pipeline, dalla versione più recente: la voce di changelog di ogni versione (src/pipelines/cross-workspace-csv-merge/changelog.ts) e, sotto, cosa è cambiato nel piano rispetto alla precedente. Vedi versioni di pipeline.

v32026-10-09 · ETL-193, ETL-195

Modifiche

• Nessun cambio di comportamento: il piano è identico alla v2. Gli import seguono l'infrastruttura (changelog, configurazione delle istanze, report di volume) uscita da `src/pipelines/` (ADR-0093 §1), e la pipeline si dichiara nel catalogo di `etl.config.ts` invece che in un registro (ADR-0093 §2)

Il piano non cambia rispetto alla v2.

v22026-10-06 · ETL-152

Aggiunte

• Variabile e origine su ogni riga dei nodi Noeva: base URL, workspace A, file e workspace B, che col default compare con origine `default` (ADR-0089)
• Riga `credenziale` sui nodi Noeva, con lo stato della chiave API e mai il valore
• Configurazione della pipeline nel piano: le variabili `NOEVA_MERGE_*` con valore e origine

Modifiche

• Senza `.env` i nodi mostrano un client Noeva d'esempio con le variabili mancanti marcate, invece di omettere base URL e workspace A
• Il workspace A è il primo che l'istanza serve: con `NOEVA_WORKSPACE_IDS` impostata, `NOEVA_WORKSPACE_ID` non conta più

Il piano rispetto alla v1:

Nodi

• ~read-workspace-aconnettore
• ~read-workspace-bconnettore
• ~workspace-aconnettore
• ~workspace-bconnettore

v12026-10-05 · ETL-120

Aggiunte

• Versione iniziale sotto versionamento (ADR-0086)

Prima versione: non c'è una versione precedente da confrontare.