Connettore: PostgreSQL

Source, Sink e resolver del nodo lookup per PostgreSQL, basati su pg + pg-cursor. Codice: src/connectors/postgres/.

Configurazione

loadPgPoolConfig(prefix?, env?) costruisce una PoolConfig dalle variabili <PREFIX>HOST|PORT|USER|PASSWORD|DATABASE (default prefisso PG) e opzionale <PREFIX>SSL=true. Usa prefissi distinti per DB sorgente e destinazione. Le credenziali stanno solo in .env (vedi .env.example).

createPgPool(config) apre il Pool di pg su quella config, senza connettersi: una pipeline non importa il driver, e ne chiude il ciclo di vita con end(). Per etl env check, istanzaPgDichiarata(id) è l'istanza da passare ad ambienteDellaPipeline nell'environment della pipeline, con utente e password che il piano non mostra.

Le variabili sono dichiarate una volta in MANIFEST_PG: cinque obbligatorie, senza default nemmeno per la porta, e SSL (default false). Con variabili mancanti il loader lancia un ConfigError che le elenca tutte, anche nel campo mancanti; una porta non valida è un ConfigError che nomina la variabile, non il valore. Le istanze passano dal registry (loadPostgresRegistry, PG_INSTANCES): l'istanza default usa PGHOST…, un'istanza src usa PG_SRC_HOST….

Nel piano, dove non c'è un pool, istanzaPgPerIlPiano(id, env) dà host, porta e database dell'istanza id con le loro variabili, o l'esempio (postgres.esempio) se l'istanza non si costruisce. La carica loadPgConnessione, che non legge le altre istanze: una rotta non trascina le sane. dettagliConnessioneSql ne fa le righe del descriptor, con variabile e origine e in fondo la credenziale (password · impostata, mai il valore). Vedi la convenzione dei descriptor.

import {
  createPgPool,
  loadPgPoolConfig,
  postgresSource,
  postgresSink,
} from "@noeva/etl-core/connectors/postgres";

const pool = createPgPool(loadPgPoolConfig("PG"));
try {
  // estrazione incrementale: il watermark lo legge e lo committa il runner
  const source = postgresSource<Order>({
    pool,
    instanceId: "erp",
    query:
      "SELECT id, amount, updated_at FROM orders WHERE updated_at >= :watermark ORDER BY updated_at",
    watermark: {
      column: "updated_at",
      type: "timestamp",
      initial: "1970-01-01 00:00:00",
    },
    batchSize: 1000,
  });

  // caricamento idempotente (upsert su chiave naturale)
  const sink = postgresSink<FactOrder>({
    pool,
    table: "dwh.fact_orders",
    columns: ["orderId", "amount", "date"],
    conflictTarget: ["orderId"],
  });
} finally {
  await pool.end(); // il chiamante possiede il ciclo di vita del Pool
}

Source — postgresSource

• Streaming via cursor server-side: legge batchSize righe alla volta, non carica l'intero result-set.
• Incrementale: con l'opzione watermark, vedi sotto. Una query parametrizzata a mano con values resta possibile, ma il watermark non lo salva nessuno.
• Rispetta ctx.signal (annullamento) tra un batch e l'altro.
• Errori di estrazione → ExtractError.

Source incrementale: watermark

postgresSource({
  pool,
  instanceId: "erp",
  query: "SELECT * FROM ordini WHERE stato = $1 AND updated_at >= :watermark",
  values: ["aperto"],
  watermark: {
    column: "updated_at", // colonna del risultato
    type: "timestamp", // "timestamp" | "integer"
    initial: "1970-01-01 00:00:00", // da dove parte la prima run
    lookback: 5 * 60_000, // opzionale, millisecondi, solo per "timestamp"
  },
});

Il runner legge lo stato di pipeline all'avvio e lo committa a run conclusa (nodo source). Quello che fa il connettore:

• Il predicato lo scrivi tu, con >= e il segnaposto :watermark. La source lo lega come ultimo parametro ($n dopo i values), quindi i $1… della query non cambiano numero. Il DAG mostra la query com'è scritta.
• Alla costruzione rifiuta una query senza :watermark (sarebbe un full scan a ogni run) e un :watermark senza l'opzione watermark. Conta solo il codice: un letterale, un commento o un cast ::watermark non sono il segnaposto.
• column va selezionata. Se non è nel risultato l'estrazione fallisce; una riga con la colonna a NULL non sposta il watermark.
• Precisione nativa. Il watermark è il testo del database, microsecondi compresi:
  • timestamp senza fuso arriva già come stringa (vedi sotto);
  • timestamptz resta una Date nel record, come sempre, ma il watermark è il testo con cui il database ha mandato il valore, con il suo offset;
  • integer: int4 e int8, quest'ultimo come stringa, quindi anche oltre 2^53.
• instanceId si salva accanto al valore: se cambia, la run fallisce invece di applicare il watermark di un database a un altro.
• Il sink a valle dev'essere idempotente: con >= le righe al confine si rileggono.

Date e orari in lettura

Source e lookup leggono con i parser di tipo del connettore (pgReadTypes(), in types.ts):

Tipo Postgres	OID	Valore nel record
date	1082	stringa YYYY-MM-DD
timestamp without time zone	1114	stringa YYYY-MM-DD HH:mm:ss[.ffffff]
date[]	1182	array di stringhe YYYY-MM-DD
timestamp without time zone[]	1115	array di stringhe YYYY-MM-DD HH:mm:ss[.ffffff]
timestamptz	1184	Date (è un istante: nessuna ambiguità di fuso)

Il default di pg trasforma i primi quattro in Date nel fuso del processo: la stessa riga letta da due macchine con TZ diversa diventerebbe due istanti, quindi due hash di riga e due chiavi di versione. La stringa nella forma nativa del database è identica ovunque.

• Schema Zod: una colonna date o timestamp si dichiara z.string(), non z.date(). ⚠️ z.coerce.date() la accetterebbe, ma la interpreta di nuovo nel fuso del processo: non usarlo su queste colonne.
• Array: un elemento NULL arriva come null, quindi lo schema è z.array(z.string().nullable()). Un array a più dimensioni arriva annidato.
• Se il valore è in realtà un istante (un timestamp che l'applicazione scrive in UTC), dichiaralo nella query: created_at AT TIME ZONE 'UTC' AS created_at lo porta a timestamptz e arriva come Date.
• I parser valgono per query, non sono registrati con pg.types.setTypeParser: nessun altro codice nel processo cambia comportamento, e funzionano anche con un Pool costruito a mano. Una lettura fatta direttamente con pool.query(...), fuori dal connettore, segue il default di pg.
• Valori speciali: infinity, -infinity e le date BC arrivano come testo, anche dentro un array. Sono valori Postgres legittimi e identici in ogni ambiente, quindi il connettore li lascia passare. Uno schema che deve escluderli usa una regex, non un z.string() nudo: così la riga va in dead-letter invece di fermare l'estrazione.
• Il connettore legge sempre in formato testo: il formato binario resta al default di pg.
• Fuori da questi quattro tipi il connettore non guarda dentro il valore: un intervallo (daterange, tsrange), un tipo composito o una data portata a testo nella query (giorno::text) arrivano come li scrive il server, e seguono il DateStyle senza verifica. Se servono in un hash, formattali nella query: to_char(giorno, 'YYYY-MM-DD').

DateStyle della sessione

La stringa è quella che manda il server, e segue il DateStyle della sessione. Il default di Postgres è ISO, che dà le forme in tabella; con un ruolo o un database configurato diversamente la stessa data arriverebbe in un'altra forma (29.03.2026), quindi con un altro hash di riga.

Il connettore verifica la forma di ogni valore letto da questi quattro tipi, e se non è ISO la lettura fallisce, senza consegnare il batch che lo contiene né spostare il watermark (i batch già consegnati restano a valle, come per ogni errore di estrazione):

• postgresSource solleva un ExtractError;
• postgresLookup solleva un LookupError non transitorio: ritentare ridarebbe la stessa forma.

Il messaggio porta il tipo, il valore e il rimedio. Si risolve sul database, una volta:

ALTER ROLE etl_reader SET datestyle = 'ISO';
-- oppure: ALTER DATABASE analytics SET datestyle = 'ISO';

Vale per le sessioni aperte dopo il comando. In alternativa, su una connessione diretta, il Pool può chiederlo all'avvio: createPgPool({ ...loadPgPoolConfig("PG"), options: "-c DateStyle=ISO" }).

Il connettore non lo impone da sé alla connessione. Un pooler in transaction mode (PgBouncer, Supabase) può ignorare il parametro di startup o rifiutare la connessione, e un SET di sessione lì non regge, perché ogni transazione può finire su un backend diverso. La verifica sui valori vale invece con qualunque Pool e qualunque pooler.

Sink — postgresSink

• Due modalità: l'upsert descritto qui (default) e la modalità SCD2, che conserva la storia.
• Idempotente: INSERT ... ON CONFLICT (conflictTarget) DO UPDATE in transazione.
  • conflictTarget vuoto → INSERT non idempotente (sconsigliato).
  • tutte le colonne nella chiave → DO NOTHING.
• Scrive a batch (batchSize, default 500); l'intera load è in un'unica transazione (atomica).
• Dichiara Sink.held (in lavorazione nei contatori dal vivo): le righe nel batch non ancora scritte, anche in SCD2. Con retry/rateLimit sul nodo conta invece il blocco del runner. Vedi sink, «Dal vivo».
• Errori di caricamento → LoadError con ROLLBACK.

Nota: la transazione singola garantisce atomicità ma tiene aperto un lock sul carico molto grande; per volumi enormi valutare commit a finestre (futuro ADR).

Sink in modalità SCD2: dimensioni storicizzate

Con mode: "scd2" il sink non sovrascrive: quando un'entità cambia chiude la versione corrente e ne apre una nuova, alla data di riferimento della run. È il tPostgresqlSCD di Talend. La logica è comune a SQL Server e vive in src/connectors/shared/scd2/.

const sink = postgresSink<ClienteRiga>({
  pool,
  table: "dwh.dim_cliente",
  mode: "scd2",
  // solo ciò che arriva nel record: chiave naturale, chiave durevole, business, hash
  columns: ["codice", "cliente_sk", "nome", "citta", "riga_hash"],
  naturalKey: ["codice"],
  hashColumn: "riga_hash",
  validFrom: "valid_from",
  validTo: "valid_to",
  isCurrent: "is_current", // opzionale
  versionKey: { column: "versione_sk", entity: "cliente" },
  executionIdColumn: "execution_id",
});

Opzione	Obbligatoria	Significato
mode	sì	"scd2". Vieta conflictTarget a livello di tipo
naturalKey	sì	Colonne della chiave naturale, dentro columns. L'ordine entra nella chiave di versione
hashColumn	sì	Colonna dell'hash di riga calcolato da hashRow a monte, dentro columns
validFrom	sì	Colonna dell'inizio di validità. La scrive il sink
validTo	sì	Colonna della fine di validità, esclusa. NULL sulla versione corrente
isCurrent	no	Colonna booleana «è la versione corrente», mantenuta dal sink
versionKey	sì	{ column, entity }: colonna della chiave di versione ed entity della chiave surrogata
executionIdColumn	sì	Colonna con l'executionId dell'esecuzione che ha aperto, corretto o ricalcolato la versione

Le colonne scritte dal sink non possono stare in columns, e non possono ripetersi: sono errori di configurazione alla costruzione, come una naturalKey vuota o fuori da columns.

I due avvisi che il sink non può verificare

⚠️ hashRow deve coprire esattamente le colonne di business scritte. L'hash è opaco: il sink lo confronta, non lo ricalcola. Una colonna scritta ma non tracciata cambia senza aprire una versione; una colonna tracciata ma non scritta apre versioni identiche a vista. La chiave naturale, la chiave durevole e l'hash stesso restano fuori da hashRow.

⚠️ versionKey.entity deve essere la stessa entity della surrogateKey a monte. Con due entity diverse la chiave durevole e la chiave di versione vivono in namespace diversi, e una seconda pipeline sulla stessa dimensione produrrebbe chiavi di versione incompatibili.

Parametri della run

Parametro	Valore	Effetto
referenceDate	stringa YYYY-MM-DD, obbligatoria	È l'inizio di validità delle versioni aperte. Nessun fallback a now()
scd2Rehash	booleano, opzionale	true trasforma la run in un rehash

Una referenceDate assente, non stringa o che non è una data di calendario (2026-02-30) fa fallire il load prima di aprire la transazione e di consumare un record.

pnpm run etl run dim-cliente --params '{"referenceDate":"2026-09-30"}'

Cosa succede a un record

D è la data di riferimento.

Versione corrente	Hash	Esito	Scrittura
assente	—	nuova	inserimento, con inizio di validità D
presente	uguale	invariata	nessuna
aperta prima di D	diverso	cambiata	chiusura della corrente a D + inserimento
aperta in D	diverso	correzione in giornata	aggiornamento sul posto
aperta dopo D	diverso	errore	nessuna: rollback

• La chiave di versione è deriveSurrogateKey(entity, [...chiaveNaturale, D]): deterministica, quindi rieseguire la stessa data non la cambia.
• Una correzione in giornata aggiorna colonne di business, hash ed executionId; chiave di versione e validità restano. Il sink non cancella mai righe, nemmeno quando la correzione riporta i valori uguali alla versione precedente: una versione può già essere referenziata da un fatto.
• Rieseguire la stessa data con gli stessi dati non scrive niente: written: 0.
• Chiusure e aperture viaggiano in uno statement per batch; le correzioni in giornata in uno statement per record. Un rerun sulla stessa data dopo un cambio massivo a monte è quindi più lento di un carico normale.
• La correzione in giornata vale per un'esecuzione successiva. Se la versione di oggi l'ha aperta o corretta questa stessa esecuzione, in un batch o in un blocco precedente, la stessa chiave è arrivata due volte con valori diversi: è l'errore di chiave duplicata.

DDL di riferimento

Il sink non crea niente. La tabella si prepara così:

CREATE TABLE dwh.dim_cliente (
  versione_sk  uuid PRIMARY KEY,           -- chiave di versione, scritta dal sink
  cliente_sk   uuid NOT NULL,              -- chiave durevole, da surrogateKey
  codice       text NOT NULL,              -- chiave naturale
  nome         text NOT NULL,
  citta        text NOT NULL,
  riga_hash    char(67) NOT NULL,          -- "v1:" + 64 caratteri esadecimali
  valid_from   date NOT NULL,
  valid_to     date,                       -- NULL = versione corrente
  is_current   boolean NOT NULL,           -- solo se si dichiara isCurrent
  execution_id text NOT NULL,
  CHECK (valid_to IS NULL OR valid_to > valid_from)
);

-- al più una versione corrente per entità
CREATE UNIQUE INDEX dim_cliente_corrente
  ON dwh.dim_cliente (codice) WHERE valid_to IS NULL;

-- al più una versione al giorno, e ricerche per data
CREATE UNIQUE INDEX dim_cliente_storia
  ON dwh.dim_cliente (codice, valid_from);

Un vincolo EXCLUDE che vieti gli intervalli sovrapposti è opzionale: i due indici univoci coprono ciò che il sink scrive. Una violazione dell'indice sulla versione corrente vuol dire che qualcosa ha scritto sulla tabella fuori dal sink.

I fatti puntano alla chiave durevole per «l'entità» e alla chiave di versione per «l'entità com'era quel giorno». Per trovare la versione valida a una data: valid_from <= data AND (valid_to IS NULL OR data < valid_to).

Errori

Tutti annullano l'intero load. Nessuno è transitorio, tranne il lock occupato.

Errore	Cosa fare
Hash assente o non nel formato vN:<hex>	Manca hashRow a monte, o hashColumn punta alla colonna sbagliata
Componente della chiave naturale null o assente	Lo intercetta surrogateKey a monte, con la sua dead-letter
Versione dell'hash diversa fra ingresso e versione corrente	Rehash
Data di riferimento precedente, su un record cambiato	Il backfill fuori ordine non è supportato. Un record invariato passa
Chiave naturale duplicata con valori diversi	dedup a monte
Violazione di un indice univoco	Qualcosa ha scritto sulla tabella fuori dal sink, o due chiavi in arrivo sono uguali per il database
Una chiave naturale ha più di una versione corrente	Manca l'indice univoco filtrato del DDL di riferimento
Chiavi diverse in arrivo corrispondono alla stessa versione	Maiuscole o spazi che la collation ignora: normalizzare la chiave a monte
Un altro caricamento sta scrivendo sulla tabella (transitorio)	retry sul nodo, o rilanciare la run

Il sink non è isolabile per record: non ha loadOne, quindi sotto un try-catch un errore arriva al ramo catch senza raw.

Concorrenza e retry

• Ogni load prende un lock applicativo per tabella di transazione (pg_try_advisory_xact_lock), senza aspettare: due caricamenti sulla stessa dimensione non si accavallano, e il secondo fallisce con un errore transitorio. Il lock è legato al nome della tabella così com'è scritto in table: dwh.dim_cliente e dim_cliente sono due lock.
• Il sink è retrySafe. Con retry sul nodo un blocco ritentato non apre versioni doppie, perché il rollback lo ha annullato. Dopo un errore definitivo la dimensione è caricata a metà per quella data: si riesegue la stessa data di riferimento, e i blocchi già confermati risultano invariati.
• Chi vuole l'atomicità dell'intera dimensione non mette retry sul nodo. Senza retry il lock è tenuto per tutta la run.

Nel report

written è il numero delle righe scritte: versioni aperte più correzioni. Il nodo riporta anche versions: { opened, closed, corrected, unchanged } (vedi GraphRunReport). L'executionId non è nel report: lo riporta il log sink.scd2.load, accanto al runId.

Lookup — postgresLookup

Alimenta il nodo lookup del grafo: risolve un batch di chiavi con un solo round-trip, in sola lettura. È il gemello di mssqlLookup: stessa query, stesso controllo di sola lettura, stessa normalizzazione — la parte comune vive in src/connectors/shared/sql-lookup.ts, qui cambia solo il driver. Unica differenza: l'opzione keysAs: "rows" (chiavi come tabella) oggi esiste solo su mssqlLookup.

postgresLookup({
  pool,
  instanceId: "dwh",
  query:
    "SELECT UPPER(codice) AS chiave, ordine FROM v WHERE UPPER(codice) IN (@chiavi)",
  keyColumn: "chiave",
});

• Il segnaposto @chiavi (KEYS_PLACEHOLDER) è lo stesso di SQL Server e viene espanso in un parametro posizionale per chiave ($1, $2, …): i valori restano bindati. Con zero chiavi diventa un insieme vuoto valido, non SQL invalido.
• keyColumn dice quale colonna del result set riassocia le righe alle chiavi. Una colonna chiave assente, o non scalare, è un LookupError — non zero risultati.
• Sola lettura imposta a build(): assertReadOnlyQuery rifiuta tutto ciò che non è una singola SELECT/WITH — anche una CTE che scrive (WITH d AS (DELETE … RETURNING …)) e un SELECT … FOR UPDATE.

⚠️ Maiuscole e spazi. Le chiavi arrivano al DB normalizzate (trim + maiuscolo) e senza duplicati; le righe vengono riassociate con la stessa normalizzazione e restituite sulla chiave così come l'ha passata il nodo. Su SQL Server la collation tipica è case-insensitive e ignora gli spazi in coda, quindi WHERE codice IN (@chiavi) basta; su Postgres no: se la colonna può contenere minuscole o spazi, confronta su UPPER(TRIM(codice)) (o su un indice espressione equivalente), altrimenti quelle righe non vengono trovate — senza errore, come un non-trovato.

⚠️ È un controllo sintattico, non un modello di sicurezza: non vede, per esempio, una funzione con effetti collaterali dentro la SELECT (nextval(), set_config()). La garanzia vera che la pipeline non scriva resta l'utenza di sola lettura sul database.

Task — postgresSqlTask

Esegue statement SQL senza flusso di record, in una transazione: è ciò che gira in un nodo task.

postgresSqlTask({
  pool,
  instanceId: "dwh",
  name: "pulisci-giorno",
  statements: ["DELETE FROM dwh.ordini_staging WHERE giorno = @giorno"],
  params: { giorno: (ctx) => ctx.params["giorno"] as string },
  destructive: true,
});

• Statement statici, uno per voce, tutti nella stessa transazione (BEGIN … COMMIT, ROLLBACK su errore): TRUNCATE e ALTER TABLE … RENAME sono transazionali, quindi uno scambio di tabelle è atomico.
• I valori entrano solo come parametri bind: @nome nello statement, dichiarato in params, e tradotto in $1, $2, …. Il segnaposto è lo stesso di mssqlSqlTask. Nel descriptor compaiono i soli nomi. Postgres non accetta parametri negli statement di utilità (TRUNCATE, ALTER): i parametri servono a DELETE, UPDATE, CALL.
• destructive va dichiarato e compare nel DAG. retrySafe vale false se non lo dichiari: solo con true il nodo ammette retry.
• Errori → TaskError; la run fallisce. Le righe toccate da ogni statement vanno nel report del nodo.
• Uno statement già partito non si interrompe con l'annullamento della run: finisce, poi la transazione viene annullata.
• Come scrivere uno scambio, e cosa fare se fallisce: runbook.

Errori transitori

Un deadlock (40P01) o una serialization failure (40001), e gli errori di rete ECONNRESET ed ETIMEDOUT, marcano l'errore del sink, del lookup e del task come transitorio (context.transient: true): un nodo con una policy di retry lo ritenta. Il sink ha già annullato la transazione, e l'upsert è idempotente. In modalità SCD2 è transitorio anche il lock della tabella occupato da un altro caricamento.

Sicurezza SQL

• I valori passano sempre come parametri $n (mai concatenati).
• Identificatori (tabella/colonne) provengono dal codice e sono validati/quotati (quoteIdentifier): una stringa non conforme solleva ConfigError.

Test

• Unit (CI): src/connectors/postgres/{query,config,lookup,types,task}.test.ts — SQL building, parsing config, resolver del lookup e task con pool finto, parser di tipo. La parte comune del lookup è testata in src/connectors/shared/sql-lookup.test.ts, quella del task in src/connectors/shared/sql-task.test.ts.
• Integrazione: tests/integration/postgres.integration.test.ts — upsert idempotente, streaming e nodo lookup in un grafo contro un Postgres reale; la stessa riga letta con TZ=UTC e TZ=America/New_York dà valori identici, array date[] e timestamp[] compresi; con DateStyle non ISO sulla sessione source e lookup falliscono con un errore esplicito.
• Modalità SCD2: unit in src/connectors/shared/scd2/*.test.ts (configurazione, parametri, classificazione, SQL dei due dialetti, orchestrazione con un database finto); integrazione in tests/integration/scd2.integration.test.ts, con le stesse asserzioni su Postgres e SQL Server.
• Integrazione del task: tests/integration/sql-task.integration.test.ts — truncate, caricamento e scambio, atomicità di uno scambio che fallisce a metà, riesecuzione, parametri bind. Svuota e scambia tabelle, quindi gira solo se PGHOST è la macchina locale e con ETL_TEST_DESTRUCTIVE=1.

# richiede un Postgres raggiungibile via env PG* (vedi .env)
pnpm run test:integration