> ## Documentation Index > Fetch the complete documentation index at: https://docs.kaireonai.com/llms.txt > Use this file to discover all available pages before exploring further. # Loading Modes, Hooks & Validation > Six target load modes, four hook types, five dataset validators, in-memory row-level rule enforcement, and per-pipeline DLQ tables. Phase 4 closes the four areas the Phase 1 runtime explicitly stubbed: * All six target load modes run today; `cdc_mirror` applies an op-column-driven change feed from the batch rows (see the `cdc_mirror` row and the CDC mirror details section below). * Pre/post hooks fire around the load. * Dataset-level validators run as parameterized SQL probes. * Row-level rules evaluate in memory, with optional DLQ quarantine. ## Target load modes | Mode | When to use | What runs | | ----------------------- | ------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `append` | Insert new rows alongside existing data | Explicit-column projection from the staging rows into the target table (column-aware projection that nulls out empty strings) | | `truncate` | Replace all rows on every run | Truncates the target table and reloads from staging — **wrapped in a single Postgres transaction** so a failed reload rolls back the truncate. Aborts before any destructive statement runs when upstream produced 0 rows (see Empty-source guard below). | | `upsert` | Idempotent merge by primary-like key | `INSERT … ON CONFLICT (key) DO UPDATE SET nonkey = EXCLUDED.nonkey` (requires `upsertKey`; throws if every projected column is in `upsertKey` since there's nothing to update) | | `blue_green` | Atomic swap with zero downtime | `CREATE _new (LIKE
INCLUDING ALL)`, then the same column-aware projection used by `append`/`truncate`/`upsert` inserts into `
_new`, then a 3-step rename triple swaps it in and drops `
_old`. Aborts before the load when upstream produced 0 rows. | | `incremental_watermark` | Append-only with high-watermark filter | Reads the persisted high-water from `pipeline_watermarks`, INSERT-and-watermark-upsert wrapped in a single transaction so a failed load rolls back the watermark advance (requires `watermarkColumn`). Timestamp watermarks are stored and compared in **UTC** (serialized in-database), so the checkpoint is correct regardless of the worker's Postgres session timezone. | | `cdc_mirror` | Apply a change feed (CDC extract carrying an op column) | Op-column-driven apply in one transaction: rows whose `cdcOpColumn` value is `insert`/`update`/`upsert`/`I`/`U`/`c`/`r` (case-insensitive) upsert via `INSERT … ON CONFLICT (upsertKey) DO UPDATE`; rows with `delete`/`D` delete by key. Unknown op values fail the whole run before any DML. Requires `cdcOpColumn` + `upsertKey` (with a unique constraint on the key columns). The op column is never written to the target. | ### Empty-source guard When `loadMode` is `truncate` or `blue_green` and the upstream produced 0 rows, the runtime aborts the load **before** issuing the destructive statement. The run fails with: ``` target .
: source is empty (0 rows from ); aborting to prevent data loss. Set failOnEmptySource:false to override. ``` This default closes a real footgun: if your scheduled file lands empty or the source connector has a glitch, your live table doesn't get wiped to zero rows. Existing pipelines that genuinely want "empty file means clear the table" set `failOnEmptySource: false` on the target node. A side-effect: when an `onMissAction: alert` source comes up empty, the source executor also emits a `warning` alert into the [System Health](/operations-reporting/system-health) feed so operators see it in the topbar widget without having to open the runs page. ### Mode-switch validation Picking a load mode in the UI surfaces inline errors before save when: * `upsert` is chosen without an `upsertKey`, or with keys not in the destination schema, or with keys covering every column. * `incremental_watermark` is chosen without a `watermarkColumn`, or the column doesn't exist in the destination schema. * `cdc_mirror` is chosen without a `cdcOpColumn`, or without at least one CDC key column in `upsertKey`, or with key columns not in the destination schema. * `truncate` is chosen on a "full-refresh-shaped" pipeline (exactly one source + one target). Soft warning recommends `blue_green`; you can keep `truncate` if intentional. `parsePipelineIR` enforces the **presence** preconditions server-side (`upsert` → `upsertKey`, `incremental_watermark` → `watermarkColumn`, `cdc_mirror` → `cdcOpColumn` + non-empty `upsertKey`) so the JSON IR tab can't bypass them. The richer schema-aware checks (keys/columns must exist in the destination, `truncate` full-refresh warning) are editor-side; the "every column is in `upsertKey`" case is additionally caught at load time by the SQL builder, which refuses to emit a no-op `DO NOTHING`. ### CDC mirror details `cdc_mirror` treats the batch rows as a change feed instead of a plain load. Two fields on the target node configure it: * `cdcOpColumn` — the source column carrying the per-row change op. * `upsertKey` — the CDC key columns (the same field `upsert` mode uses). The target table needs a unique constraint or index on these columns for the `ON CONFLICT` upsert path. Op vocabulary (case-insensitive, whitespace-trimmed): | Op value | Action | | ------------------------------------------------ | ----------------------------------------------------------------------------------------------------------------------- | | `insert`, `update`, `upsert`, `I`, `U`, `c`, `r` | `INSERT … ON CONFLICT (upsertKey) DO UPDATE` — Debezium's `c` (create), `r` (snapshot read), `u` (update) map onto this | | `delete`, `D` | `DELETE` by key — matches the batch's delete keys as a tuple `IN` | Semantics: * **Fail-closed on unknown ops** — any other op value (including empty/NULL) fails the run before any DML touches the target. Silently skipping an unrecognized op would desynchronize the mirror. * **One transaction, upserts before deletes** — a key that appears as both insert/update and delete in the same batch nets out deleted. Two upsert rows for the same key in one batch fail loudly (Postgres's "cannot affect row a second time") — one op per key per batch is the contract. * **The op column is control metadata** — it is never written to the target, even when the target happens to carry a same-named column. * **Typed landing** — the same column-aware NULLIF + `::cast` projection as the other load modes. * `rowsLoaded` reports the truthful applied count (upserts + deletes). ### Blue-green details Blue-green loads the new dataset into a sibling `
_new`, runs the post-hook (e.g., a stats-collection command or a materialized-view refresh), then issues three sequential `alter table … rename` statements: ``` target → target_old target_new → target DROP target_old ``` If the INSERT into `
_new` fails, `
_new` is dropped and the upstream error rethrows so the existing `target` is untouched. ### Incremental-watermark details The high-water mark lives in the `pipeline_watermarks` table keyed by `(tenantId, pipelineId, targetSchema, targetTable)`. Each run: 1. Reads the persisted `watermarkValue` (falls back to `SELECT MAX() FROM ` on the first run after an upgrade so we don't replay history). 2. Computes the new MAX from the staging table. 3. Wraps the `INSERT … WHERE col > $1` and the watermark `upsert` in a single Postgres transaction so a failed INSERT rolls back the watermark advance. When both the target and source are empty, the watermark defaults to `0001-01-01` (timestamp) or `0` (integer/bigint) so the first non-empty run loads everything. For **timestamp** watermark columns the persisted `watermarkValue` is stored in ISO-8601 form, so it round-trips cleanly through the `::TIMESTAMPTZ` cast on the next run's `INSERT … WHERE col > $1`. (Earlier builds persisted a locale-formatted date string that Postgres rejected on the second run, so a timestamp-watermark pipeline would fail after its first successful load — now covered by a real-DB regression test.) ### Pre-load backup (optional) Add `backupBeforeLoad` to a **destructive-mode** target (`truncate` or `blue_green`) to snapshot the current target rows before the load: ```json theme={null} "backupBeforeLoad": { "enabled": true, "retainCount": 3 } ``` Before the load the runtime runs `CREATE TABLE .
_backup_ AS TABLE .
`. If that snapshot fails, the run aborts **before** the destructive statement runs — better to fail than to run an unbacked destructive load. After a successful load, backups beyond `retainCount` (default 3; newest kept, ordered by creation time rather than name) are dropped. On a **failed** load the backup table is left in place so you can recover manually with `INSERT INTO
SELECT * FROM `. The flag is ignored on non-destructive modes (`append` / `upsert` / `incremental_watermark`). ### Row-count anomaly detection (optional) Add `expectedRowCountDelta` to a target node to receive a warning alert when today's load is wildly outside the recent average: ```json theme={null} "expectedRowCountDelta": { "minPct": -20, // alarm when today < 80% of recent avg "maxPct": 200, // alarm when today > 3× recent avg "windowRuns": 7 } ``` The runtime reads the last `windowRuns` successful runs (default 7), computes the running mean of `rowsProcessed`, and emits a `warning` alert into the System Health feed if today's load is outside the band. The run still completes — anomaly detection never fails the run on its own. Skipped when there are fewer than 3 prior successful runs. **The checkpoint never moves backwards.** A batch whose maximum sits at or below the persisted watermark holds the checkpoint rather than regressing it, and raises a run warning naming both values. Before 2026-08-15 a late-arriving backfill walked the watermark backwards, and the next ordinary batch re-inserted every row between the two values — silent duplication, since `ds_` tables carry only a surrogate `BIGSERIAL` key by default and the gated INSERT has no `ON CONFLICT`. Note also that the gate is strict (`WHERE col > watermark`), so rows sharing the exact maximum value that arrive in a **later** batch are never loaded. Prefer a strictly-increasing watermark column; a date-granularity column makes ties likely. ## Hooks The IR `preHook` and `postHook` slots accept four hook kinds. The pre-hook runs **before** the load; the post-hook runs **only on successful load**. ```ts theme={null} type Hook = | { type: "sql", statement: string } // admin SQL only — no DML | { type: "refresh_mat_view", view: string } // REFRESH MATERIALIZED VIEW | { type: "webhook", url: string } // https-only POST via SSRF guard | { type: "custom_function", function: string } // Phase 5 stub ``` ### SQL hook safety Hook SQL is admin-grade. The runtime forbids any statement starting with a data-modifying verb (`drop`, `delete`, `update`, `truncate`, `insert`, `merge`, `copy`, `grant`, `revoke`, `alter`). Allowed verbs include `analyze`, `refresh`, `cluster`, `vacuum`, `set`, `explain`, etc. Embedded `;` and SQL comments are also rejected. ### Webhook safety Webhook URLs flow through `lib/security/url-validator.validateAndResolve` — the same SSRF guard the rest of the platform uses. Private IP ranges and DNS rebinding attempts are blocked before the fetch fires. ## Dataset validators ```json theme={null} "datasetLevel": { "rowCount": { "min": 100, "max": 1000000, "onFail": "abort" }, "freshness": { "withinHours": 24, "column": "updated_at", "onFail": "warn" }, "fkIntegrity": { "column": "customer_id", "refTable": "retailco.customers", "refColumn": "id", "onFail": "abort" }, "cardinality": { "column": "country", "minDistinct": 5, "onFail": "warn" }, "duplicateKey": { "columns": ["customer_id", "order_date"], "onFail": "abort" } } ``` Each validator runs a parameterized SQL probe and returns a structured `{ ok, message?, observed? }` result. The validate executor aggregates them and applies each check's `onFail`: | `onFail` | Behavior | | -------- | ------------------------------------------------------------------------------------------------------------------------------- | | `abort` | Throws; the run fails | | `warn` | Pushes a string into the `warnings` array; downstream still runs | | `skip` | Same as `warn` at dataset level — the check has no individual rows to drop, so the violation is recorded and the load continues | Every dataset check reports what it observed on the run trace under `meta.dataset.` as `{ ok, observed, message? }`, whether or not it passed. A configured check that leaves no trace entry did not run. ## Row-level rules ```json theme={null} "rowLevel": [ { "type": "notNull", "field": "id", "onFail": "abort" }, { "type": "regex", "field": "email", "pattern": "^.+@.+$", "onFail": "skip" }, { "type": "range", "field": "age", "min": 0, "max": 150, "onFail": "warn" }, { "type": "maxLength", "field": "name", "max": 200, "onFail": "skip" }, { "type": "fieldType", "field": "id", "expected": "uuid", "onFail": "abort" } ] ``` Row-level rules evaluate in memory against rows the upstream node produced (`upstream.meta.rows`). If the upstream node produces no in-memory rows (a pipeline shape that only materializes to staging), row-level rules are **skipped and a warning is recorded on the run trace** — they are never silently treated as passing. Each rule's `onFail`: | `onFail` | Behavior | | -------- | ------------------------------------------------------------- | | `abort` | Throws on first violation | | `warn` | Logs; row continues downstream | | `skip` | Drops the row; recorded in `failed[]` for optional DLQ writes | `fieldType.expected` supports: `string`, `integer`, `float`, `boolean`, `date`, `timestamp`, `json`, `uuid`. **Blank values fail numeric rules.** `range` and `fieldType: integer | float` accept a number or a numeric string and nothing else. An empty CSV cell, a whitespace-only cell, `null`, a boolean, and an absent field are all treated the same way — they are not numbers, so they violate the rule and `onFail` applies. Use `notNull` when a missing value is the condition you want to catch, and `range` when you want to bound a value that is present. ## DLQ (quarantine) ```json theme={null} "quarantine": { "enabled": true, "table": "dlq.orders" } ``` When enabled, every row that violates a `skip`- or `warn`-mode rule is written to a quarantine table. Note that a `warn`-mode row is written to quarantine **and** continues downstream to the target — quarantine is a record of what was flagged, not a diversion. **The `table` you configure is an input to the physical name, not the name itself.** The runtime derives `dlq__`, dropping any schema qualifier, so a pipeline author cannot create or write tables elsewhere in the database. `"table": "dlq.orders"` in the tenant whose id begins `cdb967a5` writes to `dlq_cdb967a5_orders`. The run trace reports the derived name as `meta.quarantineTable` — read that rather than guessing from your config. The table is created idempotently on the first failed-row write: ```sql theme={null} CREATE TABLE IF NOT EXISTS "dlq_cdb967a5_orders" ( id BIGSERIAL PRIMARY KEY, "pipelineId" TEXT NOT NULL, "runId" TEXT NOT NULL, "nodeId" TEXT NOT NULL, "ruleId" TEXT NOT NULL, "row" JSONB NOT NULL, "createdAt" TIMESTAMPTZ NOT NULL DEFAULT NOW() ); ``` The MCP `inspectFlowError` tool reads this table; the future "Errors" UI will render the same data row by row. ## Honest residuals * **`custom_function` hook implementation** — requires the plugin SDK. * **Cross-load-mode transactions across hooks** — `truncate`, `incremental_watermark`, `append`, `upsert`, and `cdc_mirror` now run their destructive + INSERT statements inside `prisma.$transaction`. Pre/post hooks still run in their own sessions, so a hook failing after a successful load won't roll back the load. ## Related * [Pipeline IR](/data/transforms/pipeline-ir) — the typed schema target/validate executors operate on. * [File Ingestion](/data/connectors/file-ingestion) — Phase 3 source-side companion. * [MCP Flow Server](/ai-ml/mcp-flow-server) — `inspectFlowError` reads the DLQ table.