Connettore: SQL Server

Source e Sink per Microsoft SQL Server, basati sul driver mssql (tedious, puro JS). Codice: src/connectors/sqlserver/.

Configurazione

loadMssqlConfig(prefix?, env?) costruisce la config mssql dalle variabili <PREFIX>HOST|PORT|USER|PASSWORD|DATABASE (default prefisso MSSQL_) e opzionali <PREFIX>ENCRYPT / <PREFIX>TRUSTCERT. Default dev (cert self-signed): encrypt=true, trustServerCertificate=true. In produzione usare MSSQL_TRUSTCERT=false con un certificato valido. Credenziali solo in .env (vedi .env.example).

Le variabili sono dichiarate una volta in MANIFEST_MSSQL: cinque obbligatorie, senza default nemmeno per la porta, ed ENCRYPT / TRUSTCERT (default true, solo false li spegne). Con variabili mancanti il loader lancia un ConfigError che le elenca tutte, anche nel campo mancanti. Le istanze passano dal registry (loadMssqlRegistry, MSSQL_INSTANCES): la default usa MSSQL_HOST…, la dwh usa MSSQL_DWH_HOST….

createMssqlPool(config, { requestTimeout? }) apre il ConnectionPool di mssql su quella config, senza connettersi: una pipeline non importa il driver, si connette con connect() e ne chiude il ciclo di vita con close(). requestTimeout (ms) alza il tempo massimo di una richiesta oltre i 15 s del driver, per una query che lo richiede; la config dell'istanza non cambia. Per etl env check, istanzaMssqlDichiarata(id) è l'istanza da passare ad ambienteDellaPipeline nell'environment della pipeline, con utente e password che il piano non mostra.

Nel piano, dove non c'è un pool, istanzaMssqlPerIlPiano(id, env) dà host, porta e database dell'istanza id con le loro variabili, o l'esempio (sqlserver.esempio) se l'istanza non si costruisce. La carica loadMssqlConnessione, che non legge le altre istanze: una dwh incompleta non porta all'esempio la default. dettagliConnessioneSql ne fa le righe del descriptor, anche per i lookup (sqlserver-lookup), con variabile e origine e in fondo la credenziale. Vedi la convenzione dei descriptor.

import {
  createMssqlPool,
  loadMssqlConfig,
  mssqlSource,
  mssqlSink,
} from "@noeva/etl-core/connectors/sqlserver";

const pool = await createMssqlPool(loadMssqlConfig("MSSQL_")).connect();
try {
  const source = mssqlSource<Order>({
    pool,
    instanceId: "erp",
    query:
      "SELECT id, amount, updated_at FROM dbo.orders WHERE updated_at >= :watermark ORDER BY updated_at",
    watermark: {
      column: "updated_at",
      type: "timestamp",
      initial: "1970-01-01T00:00:00",
    },
    highWaterMark: 1000,
  });

  const sink = mssqlSink<FactOrder>({
    pool,
    table: "dbo.fact_orders",
    columns: ["orderId", "amount", "date"],
    conflictTarget: ["orderId"],
  });
} finally {
  await pool.close(); // il chiamante possiede il ciclo di vita del pool
}

Source — mssqlSource

• Streaming (request.stream) con backpressure via pause()/resume(): non carica l'intero result-set.
• Parametri nominali via inputs (@nome).
• Incrementale: con l'opzione watermark, vedi sotto.
• Rispetta ctx.signal (annullamento). Errori di estrazione → ExtractError.

Source incrementale: watermark

mssqlSource({
  pool,
  instanceId: "erp",
  query:
    "SELECT id, importo, updated_at FROM dbo.ordini WHERE stato = @stato AND updated_at >= :watermark",
  inputs: { stato: "aperto" },
  watermark: {
    column: "updated_at", // colonna del risultato
    type: "timestamp", // "timestamp" | "integer"
    initial: "1970-01-01T00: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 al parametro @watermark: quel nome non può stare fra gli inputs.
• 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, non i letterali né i commenti.
• column va selezionata. Se non è nel risultato l'estrazione fallisce; una riga con la colonna a NULL non sposta il watermark.
• Precisione. Un datetime2 arriva dal driver come Date, ferma al millisecondo, più la frazione restante: il watermark si ricompone fino al microsecondo, troncando (con >= troncare rilegge una riga, non la salta). Il testo è senza offset, nelle cifre UTC della Date: vale con useUTC attivo, che è il default del driver. Una colonna già testuale passa com'è.
• Colonne datetime e smalldatetime. Il watermark è legato come testo con sei cifre decimali, e SQL Server non converte in datetime una stringa con più di tre: il predicato va scritto con un cast, WHERE updated_at >= CAST(:watermark AS datetime2). Con datetime2 e datetimeoffset non serve.
• rowversion non è un tipo di watermark: si espone come intero dalla query (CONVERT(bigint, rv) AS rv) e si dichiara type: "integer". Un bigint arriva come stringa.
• Niente ORDER BY obbligatorio: il massimo si calcola su tutte le righe estratte.
• Il sink a valle dev'essere idempotente: con >= le righe al confine si rileggono.

Sink — mssqlSink

• Due modalità: l'upsert descritto qui (default) e la modalità SCD2, che conserva la storia.
• Idempotente: MERGE ... ON (conflictTarget) in transazione.
  • conflictTarget vuoto → INSERT non idempotente (sconsigliato).
  • tutte le colonne nella chiave → MERGE con solo WHEN NOT MATCHED (insert-if-absent).
• Batch (batchSize, default 200): SQL Server limita a 2100 i parametri per richiesta, quindi batchSize × numColonne deve restare sotto tale soglia.
• 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.

Sink in modalità SCD2: dimensioni storicizzate

Con mode: "scd2" il sink non fa MERGE: quando un'entità cambia chiude la versione corrente e ne apre una nuova, alla data di riferimento della run. È il tMSSqlSCD di Talend.

Opzioni, parametri della run (referenceDate, scd2Rehash), classificazione dei record, errori e report sono identici a Postgres: la logica è la stessa, in src/connectors/shared/scd2/, ed è descritta nella scheda Postgres. Qui c'è solo ciò che cambia.

const sink = mssqlSink<ClienteRiga>({
  pool,
  table: "dbo.dim_cliente",
  mode: "scd2",
  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",
});

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.

⚠️ versionKey.entity deve essere la stessa entity della surrogateKey a monte, altrimenti chiave durevole e chiave di versione vivono in namespace diversi.

DDL di riferimento

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

CREATE TABLE dbo.dim_cliente (
  versione_sk  uniqueidentifier NOT NULL       -- chiave di versione, scritta dal sink
    CONSTRAINT pk_dim_cliente PRIMARY KEY NONCLUSTERED,
  cliente_sk   uniqueidentifier NOT NULL,      -- chiave durevole, da surrogateKey
  codice       nvarchar(50) NOT NULL,          -- chiave naturale
  nome         nvarchar(200) NOT NULL,
  citta        nvarchar(200) NOT NULL,
  riga_hash    char(67) NOT NULL,              -- "v1:" + 64 caratteri esadecimali
  valid_from   date NOT NULL,
  valid_to     date NULL,                      -- NULL = versione corrente
  is_current   bit NOT NULL,                   -- solo se si dichiara isCurrent
  execution_id varchar(64) NOT NULL,
  CONSTRAINT ck_dim_cliente_validita CHECK (valid_to IS NULL OR valid_to > valid_from)
);

-- al più una versione al giorno; clustered: tiene vicine le versioni della stessa entità
CREATE UNIQUE CLUSTERED INDEX dim_cliente_storia
  ON dbo.dim_cliente (codice, valid_from);

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

• La PK sulla chiave di versione è non clustered: un UUIDv5 è distribuito a caso, e come indice clustered frammenterebbe la tabella.
• Le colonne di validità sono date, mai datetime: con una lingua di sessione non inglese SQL Server legge '2026-03-04' come 3 aprile in un datetime, mentre in un date la forma YYYY-MM-DD è sempre letta allo stesso modo.
• execution_id può essere varchar o uniqueidentifier: il sink confronta l'identificativo senza badare alle maiuscole.
• SQL Server non ha un vincolo contro gli intervalli sovrapposti: i due indici univoci coprono ciò che il sink scrive.

Cosa cambia rispetto a Postgres

• Lock applicativo: sp_getapplock di transazione con timeout zero, sulla risorsa noeva-etl:scd2:<table>. Il nome è quello scritto in table: dbo.dim_cliente e dim_cliente sono due lock.
• Parametri per richiesta: vale il limite di 2100, e in modalità SCD2 ogni riga usa numColonne + 1 parametri (o colonneDellaChiave + 1, se di più). Senza batchSize il sink sceglie il batch più grande che ci sta, fino a 200; un batchSize esplicito troppo grande è un ConfigError alla costruzione.
• Collation: con una collation case-insensitive "ABC" e "abc" sono la stessa riga per il database ma due chiavi diverse per il sink, e lo stesso vale per uno spazio in coda. La chiave naturale deve arrivare al sink già in forma canonica, come per surrogateKey. Il sink se ne accorge quando le due forme arrivano nello stesso batch (errore «corrispondono alla stessa versione») o sono entrambe nuove (violazione dell'indice univoco). ⚠️ Se arrivano in batch diversi non può accorgersene: la versione corrente rimbalza fra le due forme, una versione nuova a ogni data.

Task — mssqlSqlTask

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

mssqlSqlTask({
  pool,
  instanceId: "dwh",
  name: "svuota-staging",
  statements: ["TRUNCATE TABLE dbo.ordini_staging"],
  destructive: true,
  retrySafe: true,
});

• Statement statici, uno per voce, tutti nella stessa transazione: TRUNCATE TABLE, sp_rename e ALTER SCHEMA … TRANSFER sono transazionali, quindi uno scambio di tabelle è atomico.
• I valori entrano solo come parametri bind: @nome nello statement, dichiarato in params. Nel descriptor compaiono i soli nomi.
• destructive va dichiarato e compare nel DAG. retrySafe vale false se non lo dichiari: solo con true il nodo ammette retry.
• Errori → TaskError con ROLLBACK; la run fallisce. Le righe toccate vanno nel report del nodo.
• La parte indipendente dal DBMS è condivisa con postgresSqlTask e vive in src/connectors/shared/sql-task.ts. Come scrivere uno scambio, e cosa fare se fallisce: runbook.

Lookup — mssqlLookup

Alimenta il nodo lookup del grafo: risolve un batch di chiavi con un solo round-trip, in sola lettura.

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

La parte indipendente dal DBMS (controllo di sola lettura, segnaposto, normalizzazione, raggruppamento) è condivisa con postgresLookup e vive in src/connectors/shared/sql-lookup.ts: la stessa query funziona sui due database.

• Il segnaposto @chiavi (KEYS_PLACEHOLDER) viene espanso in un parametro per chiave (@k0, @k1, …): si generano solo i nomi, i valori restano bindati. Con zero chiavi diventa un insieme vuoto valido, non SQL invalido.

• keysAs (default "list") sceglie la forma del segnaposto. Con "list" diventa @k0, @k1 per una IN (@chiavi). Con "rows" diventa (VALUES (@k0), (@k1)) e la query lo usa come tabella (FROM @chiavi AS k(chiave) JOIN v ON …): serve quando il confronto non è un'uguaglianza, per esempio «la chiave contiene questa colonna». Costa una scansione del lato destro, quindi va usata su poche chiavi. Con zero chiavi il segnaposto diventa un rowset vuoto a colonna nvarchar.

  In entrambe le forme @chiavi va scritto una volta sola: viene sostituito solo alla prima occorrenza.

• keyColumn dice quale colonna del result set riassocia le righe alle chiavi. La chiave è normalizzata (trim + maiuscolo) sui due lati, così un UPPER() dimenticato nella vista non produce zero match in silenzio. Al DB partono le chiavi normalizzate senza duplicati; le righe tornano indicizzate per la chiave così come l'ha passata il nodo (" a" riceve le righe di A). 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 (niente INSERT/UPDATE/DELETE/ MERGE/EXEC/INTO, niente ; multipli). Commenti e stringhe letterali sono rimossi prima dell'analisi, così WHERE nota = 'da aggiornare (update)' passa e una parola vietata dentro un commento no.

⚠️ È un controllo sintattico, non un modello di sicurezza: la garanzia vera che la pipeline non scriva resta l'utenza di sola lettura sul database. Serve a trasformare un errore di configurazione in un fallimento immediato e leggibile.

Errori transitori

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

Sicurezza SQL

• I valori sono sempre parametri (@p0, @p1, ...), mai concatenati.
• Identificatori (tabella/colonne) dal codice, validati e quotati con [...] (quoteIdentifier); stringa non conforme → ConfigError.

Ambiente di test

SQL Server si prova in Docker con l'immagine ufficiale, che è solo linux/amd64: su Apple Silicon gira in emulazione.

Test

• Unit (CI): src/connectors/sqlserver/{query,config,lookup,task}.test.ts — MERGE building, parsing config, resolver del lookup e task con pool finto. 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/mssql.integration.test.ts — MERGE idempotente e streaming contro il container reale.
• Modalità SCD2: unit in src/connectors/shared/scd2/*.test.ts; 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 con sp_rename, atomicità, parametri bind. Svuota e scambia tabelle, quindi gira solo se MSSQL_HOST è la macchina locale e con ETL_TEST_DESTRUCTIVE=1.

docker compose up -d           # avvia SQL Server di test
pnpm run test:integration      # esegue i test di integrazione (auto-skip senza env)