Skip to main content
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

​
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:
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 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): 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 <table>_new, runs the post-hook (e.g., a stats-collection command or a materialized-view refresh), then issues three sequential alter table … rename statements:
If the INSERT into <table>_new fails, <table>_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(<watermarkColumn>) FROM <target> 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:
Before the load the runtime runs CREATE TABLE <schema>.<table>_backup_<runId> AS TABLE <schema>.<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 <table> SELECT * FROM <backup_table>. 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:
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.

​
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

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: Every dataset check reports what it observed on the run trace under meta.dataset.<check> as { ok, observed, message? }, whether or not it passed. A configured check that leaves no trace entry did not run.

​
Row-level rules

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: 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)

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_<tenant-prefix>_<name>, 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:
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.