Connettore: MySQL e MariaDB

Source, Sink, lookup e task per MySQL e MariaDB, sul driver mysql2 (puro JS). Un connettore solo per i due motori: provato su MySQL 8.4 e MariaDB 11. Codice: src/connectors/mysql/. Decisioni in ADR-0109.

Configurazione

loadMysqlPoolConfig(prefix?, env?) costruisce le PoolOptions di mysql2 dalle variabili <PREFIX>HOST|PORT|USER|PASSWORD|DATABASE (default prefisso MYSQL_) e opzionali <PREFIX>SSL e <PREFIX>SSL_VERIFY. Le variabili sono dichiarate una volta in MANIFEST_MYSQL: cinque obbligatorie, senza default nemmeno per la porta, e SSL (default false) e SSL_VERIFY (default true). Con variabili mancanti il loader lancia un ConfigError che le elenca tutte, anche nel campo mancanti. Charset utf8mb4, timezone: "Z".

• Senza MYSQL_SSL=true la connessione non è cifrata.
• Con MYSQL_SSL=true il certificato del server si verifica. Per un self-signed in sviluppo si scrive MYSQL_SSL_VERIFY=false, di proposito.
• Credenziali solo in .env (vedi .env.example).

Le istanze passano dal registry (loadMysqlRegistry, MYSQL_INSTANCES): la default usa MYSQL_HOST…, la dwh usa MYSQL_DWH_HOST…. Per etl env check, istanzaMysqlDichiarata(id) è l'istanza da passare ad ambienteDellaPipeline nell'environment della pipeline.

createMysqlPool(config) apre il pool di mysql2 senza connettersi: una pipeline non importa il driver, e ne chiude il ciclo di vita con end(). Ogni connessione nuova imposta la sessione su UTC (SET time_zone = '+00:00'), e multipleStatements è sempre spento.

Nel piano, dove non c'è un pool, istanzaMysqlPerIlPiano(id, env) dà host, porta e database dell'istanza id con le loro variabili, o l'esempio (mysql.esempio) se l'istanza non si costruisce; dettagliConnessioneSql ne fa le righe del descriptor. Vedi la convenzione dei descriptor.

import {
  createMysqlPool,
  loadMysqlRegistry,
  mysqlSink,
  mysqlSource,
} from "@noeva/etl-core/connectors/mysql";

const pool = createMysqlPool(loadMysqlRegistry().get("dwh"));
try {
  const source = mysqlSource<Order>({
    pool,
    instanceId: "dwh",
    query:
      "SELECT id, amount, updated_at FROM orders WHERE updated_at >= :watermark AND kind = ? ORDER BY updated_at",
    values: ["fattura"],
    watermark: {
      column: "updated_at",
      type: "timestamp",
      initial: "1970-01-01 00:00:00",
    },
  });

  const sink = mysqlSink<FactOrder>({
    pool,
    table: "dwh.fact_orders",
    columns: ["id", "amount", "updated_at"],
    conflictTarget: ["id"],
  });
  // ... graph().source(source).sink(sink).build()
} finally {
  await pool.end();
}

Cosa arriva nei record

mysql2 farebbe Date di DATE, DATETIME e TIMESTAMP nel fuso del processo, e number di BIGINT. Il connettore legge invece come testo, con un typeCast per query (vale con qualunque pool):

Tipo MySQL / MariaDB	Nel record
DATE	"2026-03-29"
DATETIME, TIMESTAMP	"2026-03-29 02:30:00.123456", precisione intera
BIGINT	"9007199254740993", nessuna perdita oltre 2^53
DECIMAL	stringa, come in mysql2
JSON	l'oggetto su MySQL; su MariaDB JSON è LONGTEXT, quindi stringa
ogni altro	il default del driver

Così la stessa riga letta da due macchine con TZ diverse è identica, e lo stesso vale per il suo hash di riga (ADR-0070). Uno zero-date (0000-00-00) arriva com'è: escluderlo spetta allo schema della pipeline. Un TIMESTAMP è nel fuso della sessione: con createMysqlPool è UTC, con un pool costruito a mano è quello del server.

Nella scrittura vale l'inverso, con paramValue: una Date si scrive in UTC (YYYY-MM-DD HH:mm:ss.mmm), un oggetto o un array come JSON (non come lista), un bigint come testo, undefined come NULL.

Source — mysqlSource

Streamma i risultati di una query: le righe arrivano dal socket con backpressure, e il result-set non è mai in memoria (batchSize è la coda dello stream, default 500). I valori dinamici sono ?, legati nell'ordine in cui compaiono; il numero di ? deve essere quello di values, e si controlla alla costruzione. I ? dentro letterali, identificatori con backtick e commenti (-- , #, /* */) non contano.

Se lo stream non si legge fino in fondo (un errore a valle, un annullo) la connessione si chiude, non torna al pool: aveva ancora righe in arrivo.

Source incrementale: watermark

Come per Postgres e SQL Server (ADR-0068): watermark: { column, type, initial } e un predicato scritto per esteso con :watermark, che si scrive dove serve, anche prima dei ? e anche più volte. Il valore è il testo del database ("2026-03-29 02:30:00.123456"), con la precisione intera. La colonna va selezionata. Senza :watermark nella query, o con :watermark senza watermark, la costruzione fallisce.

Sink — mysqlSink

INSERT … ON DUPLICATE KEY UPDATE col = VALUES(col) in batch, dentro una transazione: rieseguire non duplica. conflictTarget non è un target come in Postgres: MySQL scatta su qualunque indice univoco o chiave primaria, e conflictTarget dice quali colonne sono la chiave naturale e quindi non si aggiornano. Deve corrispondere a un indice univoco della tabella, e il connettore non può verificarlo.

• Con tutte le colonne nella chiave la riga esistente resta com'è (a = a).
• Senza conflictTarget la INSERT non è idempotente: sconsigliato.
• La tabella deve essere InnoDB: su un motore non transazionale un errore a metà lascia i batch già scritti.
• Un solo statement porta al più 65535 segnaposto: batchSize × columns non deve superarlo.
• VALUES() è deprecato da MySQL 8.0.20 (un avviso, non un errore), ma è l'unica forma che capisce anche MariaDB.
• Non c'è la modalità SCD2: per ora i sink SQL storicizzati sono Postgres e SQL Server.

Task — mysqlSqlTask

Statement statici in una transazione, valori solo come parametri @nome dichiarati in params (diventano ?). Come per gli altri task SQL (ADR-0071): niente ; in mezzo, niente BEGIN/COMMIT, destructive dichiarato.

⚠️ Il DDL non è transazionale. TRUNCATE, DROP, ALTER, CREATE e RENAME TABLE fanno un commit implicito: un errore dopo uno di loro non lo annulla. Lo scambio atomico di due tabelle è uno statement solo:

statements: [
  "RENAME TABLE dwh.ordini TO dwh.ordini_old, dwh.ordini_staging TO dwh.ordini",
],

Vedi il runbook.

Una variabile utente di MySQL (@x) non dichiarata in params resta com'è.

Lookup — mysqlLookup

Risolve un batch di chiavi con un solo round-trip, con lo stesso segnaposto @chiavi degli altri DBMS (diventa ?, ?, …): vedi il nodo lookup. La query è validata alla costruzione (una sola SELECT; INTO è vietato, quindi anche INTO OUTFILE). Con zero chiavi non si interroga il database.

Le chiavi arrivano normalizzate (trim + maiuscolo). Le collation di default di MySQL e MariaDB (*_ci) ignorano le maiuscole; con una _bin o _cs il confronto nella WHERE va normalizzato allo stesso modo (UPPER(codice)).

Errori transitori

Un LoadError o un LookupError portano transient: true (ADR-0059) per ER_LOCK_DEADLOCK, ER_LOCK_WAIT_TIMEOUT e PROTOCOL_CONNECTION_LOST, oltre a ECONNRESET e ETIMEDOUT.

Sicurezza SQL

• Gli identificatori (tabella, colonne) sono validati e quotati con i backtick; i valori viaggiano sempre come ?, mai nel testo.
• mysql2 interpola i ? lato client (serve per lo streaming): per questo ogni valore passa da paramValue, che toglie le sorprese (array come liste, oggetti come chiave = valore).
• multipleStatements è spento: un ; in una query è un errore, non due comandi.
• La garanzia che una pipeline non scriva dove non deve resta l'utenza di sola lettura sul database; il controllo sintattico del lookup non è un modello di sicurezza.

Ambiente di test

tests/integration/mysql.integration.test.ts si salta senza MYSQL_HOST e MYSQL_DATABASE, e si lancia una volta per motore (MySQL e MariaDB). Le istruzioni per i container usa e getta sono nel runbook dei database di test del repo del core (docs/runbooks/local-test-databases.md).

Test

Unitari colocati in src/connectors/mysql/*.test.ts, con un pool finto (fake-pool.ts): senza rete.