feat(materialization): incremental persist (refresh="incremental")

[girishjeswani]

Summary

Realizes the refresh="incremental" value of the persist refresh knob on the publisher build
path. Today a persisted source is always fully rebuilt (staging + atomic rename) on every run — fine
for most sources, but prohibitive when a per-row cost dominates the build (an ML.GENERATE_TEXT
classification, an embedding call, an expensive UDF): every refresh re-pays that cost for rows that
have not changed. refresh="incremental" makes such a source build once, then update in place.

Behavior

• refresh="incremental" (on #@ persist) selects the incremental path. Unset / "full" keeps
  the default full-rebuild path, unchanged.
• First run (target absent): CREATE TABLE <target> AS (<source SQL>) — the warehouse infers the
  schema; DDL is never derived from Malloy intrinsic types.
• Subsequent runs: MERGE INTO <target> USING (<source SQL>) ON <primary_key>, WHEN MATCHED THEN UPDATE + WHEN NOT MATCHED THEN INSERT. In place (no staging), and idempotent under retry
  (a re-run is a no-op on already-merged rows, unlike a raw INSERT).
• The merge key is the source's Malloy primary_key — no new persist-level unique_key knob.
  An incremental source with no primary key is a BadRequestError at build time.

Compile-safe self-reference construct

The reserved CTE __malloy_incremental_keys lets a source refer to "what I have already
materialized" without a bootstrap problem — the dbt is_incremental() + {{ this }} analog, with no
compiler change. The author writes it with an empty body (matches nothing → compiles, and seeds
everything on the first run); on an incremental run the publisher rewrites the body to SELECT <primary_key> FROM <target>. Because the filter sits inside the source SQL, an expensive step
runs only for new rows. The construct is optional (omit it → upsert every returned row). Write it in
the canonical WITH __malloy_incremental_keys AS ( ... ) form; the AS keyword is matched
case-insensitively and anchored to the CTE name (a lowercase as still hydrates).

Force full rebuild (forceFullRebuild)

A new run flag (dbt --full-refresh analog) rebuilds every incremental source through the full
staging + rename path this run. Deliberately distinct from forceRefresh: the scheduler sets
forceRefresh on every run, so overloading it would silently turn scheduled incremental refreshes
back into full rebuilds.

Backward compatibility

Additive and off by default. The current full-rebuild body is extracted verbatim as
buildFullSource; refresh unset dispatches straight to it. getSQL() is unaffected by the
annotation, so content-addressing and reuse-rebind are identical for full sources. The reuse-skip
exemption and the new validation rules fire only for incremental sources.

Validation & dialects

• Publish gate: refresh must be full | incremental; an incremental source that resolves to
  freshness fallback: "live" is rejected — whether the fallback is declared on the source or
  inherited from the package (a live serve returns the delta, not the dataset).
• The generated MERGE uses an unqualified SET target column (SET col = S.col) for
  portability: BigQuery and Postgres 15+ reject a target-alias-qualified column there; Snowflake
  accepts unqualified. Supported on BigQuery / Snowflake / Postgres 15+; DuckDB persistence is
  unsupported generally, and Postgres < 15 has no MERGE.

Testing

• New materialization_incremental.spec.ts: refreshMode, the SQL builders
  (hydrateIncrementalKeysCte incl. nested-paren balancing, case-insensitive AS, and the
  AS-substring guard; buildMergeSQL), and the buildOneSource dispatch (full unchanged /
  first-run CTAS seed / MERGE / forceFullRebuild / missing-pk error / CTE hydration).
• Policy Rules for refresh and incremental+live (package-level and source-level) in
  persistence_policy.spec.ts (also proves the compiler carries refresh through to the wire plan).
• forceFullRebuild validation in the controller spec; run-metadata shape updated.
• tsc --noEmit clean; all affected suites green.

Docs

• docs/materialization-incremental.md — the annotation surface, the self-reference construct
  (incl. a NOT EXISTS anti-join recommendation over the NULL-fragile NOT IN), build behavior, the
  escape hatch, validation, dialects, the stable-target-name assumption, and back-compat.

Notes

• Two commits: the feature, then a self-review pass (case-insensitive CTE AS matching,
  source-level live-fallback rejection, and doc hardening — NOT EXISTS guidance + the
  stable-target-name assumption).
• All commits are DCO signed-off.
• Based on main and independent of the CLI schedule work (feat(cli): materialization schedule + listing CLI, tuning skill, and docs #909) — it needs only the scheduler
  service already merged in feat(materialization): standalone scheduler, environment-scoped listing, and schedule management UI #888. For a hosted refresh cadence, pair with scope: package + hosted
  freshness (not the version-scoped cron).
• Deliberately small: it merges in place on the source's primary_key. A richer strategy (a
  separate unique_key, an event-time watermark predicate, or building into a cloned generation
  rather than in place) is possible later and nothing here forecloses it.

[sagarswamirao]

Nice feature -- the compile-safe __malloy_incremental_keys construct is a clean way to get is_incremental() semantics with no compiler change. A few things I'd want addressed before this ships, flagging early so they don't bite downstream:

1. The MERGE path has no dialect gate (blocking).
Dispatch to the incremental path keys only off the refresh= annotation (buildOneSource, materialization_service.ts:931) -- buildIncrementalSource checks primary-key and columns but never the dialect, and the publish rules in persistencePolicyWarnings (package.ts) don't gate it either. So refresh="incremental" on a MySQL / Trino / Postgres<15 (and Databricks) connection publishes cleanly and then emits a MERGE INTO ... the engine can't run:

• MySQL has no MERGE (it's INSERT ... ON DUPLICATE KEY UPDATE) -- and since MySQL is backtick-quoted the statement looks MySQL-shaped but the verb is unparseable.
• Postgres <15 has no MERGE; Trino needs connector write support.

The result is a raw warehouse syntax error surfaced late in the build (potentially after a first-run CTAS has already created a table), rather than a clear rejection. The doc's dialect table is documentation-only -- nothing enforces it, and Trino/MySQL/Databricks aren't even listed. Could you add a publish-time rule that rejects refresh="incremental" when the source dialect isn't in the supported set (BigQuery / Snowflake / Postgres)? Postgres-must-be-15+ stays a caveat since dialectName can't reveal the version, but gating the clearly-unsupported dialects at publish would turn a confusing late failure into an actionable error.

2. Non-atomic (nested/struct/array) columns are silently dropped on subsequent runs (blocking).
First run is CREATE TABLE AS, so the warehouse materializes all SELECT columns. But subsequent runs build the MERGE INSERT list from deriveColumns, which filters to isAtomicField() (build_plan.ts:80) -- and that filtered list is the only column set used for both INSERT (...) and VALUES (...) (materialization_service.ts:160-169). A BigQuery RECORD/REPEATED or Snowflake ARRAY/OBJECT/VARIANT column is therefore present in the seeded table but omitted from every MERGE: newly-merged rows get NULL/default in that column while first-run rows carry real data (silent divergence), or the INSERT fails outright if the column is NOT NULL. This hits exactly the two dialects the feature targets. Deriving the MERGE column list from the actual target-table schema, or rejecting incremental sources that emit non-atomic columns at publish time, would close it. (JSON is fine -- JSONField is atomic; only struct/array/record drop.)

3. The forceFullRebuild description in api-doc.yaml leaks implementation detail into the public spec.
api-doc.yaml:4076 reads "Escape hatch (dbt --full-refresh analog): rebuild every incremental ... via the full staging+rename path this run, instead of MERGE-ing new rows." The spec is a public contract; a consumer generating a client shouldn't need to know dbt or the internal build strategy (staging+rename, MERGE). Something like "When true, incremental persist sources are fully rebuilt this run instead of applying only new/changed rows. Use after a schema or definition change. Independent of forceRefresh." keeps the contract clean -- the dbt/staging rationale is great as a code comment.

A few smaller things worth a look, non-blocking:

• tableExists (materialization_service.ts:1126) returns false on any thrown error, so a transient/permission blip on the probe routes to the first-run CREATE TABLE AS branch. On the supported dialects that errors "already exists" (spurious failed run, not data loss), but narrowing the catch to a table-not-found error class would avoid the misclassification.
• The MERGE wraps the CTE-bearing source as USING (WITH ... SELECT ...). BigQuery and Postgres accept a parenthesized WITH; worth confirming Snowflake accepts a leading WITH inside USING(...) on a real engine before claiming Snowflake support -- the tests assert emitted text but don't execute it.
• hydrateIncrementalKeysCte anchors with a raw regex, so the reserved CTE name appearing in a comment or string literal before the real CTE would anchor the paren-scan on the wrong span and corrupt the SQL. A guard or a note in the doc would help.
• The doc lists DuckDB as unsupported, but the incremental test suite drives the path on duckdb and recent DuckDB has MERGE INTO, so it'll actually run -- worth reconciling the doc with the enforced behavior (ties into item 1 above).

[sagarswamirao]

#913 (comment)

[housejester]

Nice feature — the __malloy_incremental_keys construct is a genuinely clean way to get is_incremental() semantics with no compiler change. @sagarswamirao's review already covers what I'd block on (no dialect gate on the MERGE; non-atomic columns silently dropped after the seed; the forceFullRebuild docstring leaking into the public spec), all correct — not restating those.

A few things not in that review. They hang off one observation about the shape of the change:

refresh="incremental" flips a persisted source from a pure function of its SQL (content-addressed, fully rebuilt each run, atomically swapped) to a stateful table that's a function of run history — seeded once, then mutated in place. The target name is stable across versions and across SQL edits (selfAssignTableName, materialization_service.ts:628), but its schema is frozen at the first-run CTAS and the table is never swapped or truncated.

Four consequences worth a decision before this ships:

1. Schema/definition drift is guarded only by docs (the general case of sagar's #2).
His non-atomic-column point is one instance of a broader hazard. The target schema is fixed at the seed CTAS (materialization_service.ts:1063); every later MERGE builds its column list from the current compiled source (deriveColumns, :1073) against that frozen schema, and the target name is stable across SQL edits (:628). So any column add/rename/remove on an incremental source — on a normal run, not forceFullRebuild — yields a MERGE that references a column the target lacks (hard error) or omits one it has (new rows get NULL/default while seed-era rows carry real data → silent divergence). The doc's "run forceFullRebuild after a schema migration" plus the stable-target-name note describe the trap, but nothing detects it. Suggest comparing deriveColumns against the live target schema before the MERGE and either auto-reseeding or failing with an actionable message — code-enforced, not operator discipline.

2. The MERGE never deletes — rows dropped from the source (or newly access-restricted) linger.
buildMergeSQL emits only WHEN MATCHED/WHEN NOT MATCHED (materialization_service.ts:146-169) — no WHEN NOT MATCHED BY SOURCE THEN DELETE. A row the source stops emitting (definition narrows, an upstream row is deleted, a category becomes access-restricted) is never removed; a full rebuild would drop it. For the append-only classify use case that's intended, but as a general property the persisted set can retain rows that no longer exist in — or are no longer authorized by — the source definition, until a forceFullRebuild. Worth (a) an explicit callout in the doc's "append-only vs upsert" section, which today only frames inserts/updates, and (b) deciding whether delete-detection is in scope for v0 or an accepted, documented limitation.

3. Cost model: incremental bounds compute-per-row, not scan — and the doc oversells it.
"The per-row cost is paid once per row for its lifetime" is true for the expensive step (ML/UDF), but reads as "incremental is cheaper, period." On BigQuery every incremental run still full-scans the base source for the anti-join (no indexes), reads the target's key column for the hydrated CTE, and — per BigQuery's MERGE semantics — scans the destination in the MERGE (billed by bytes). Because the MERGE never deletes (#2), that destination scan grows with the accumulated table each run. For a scan-dominated source, refresh="incremental" can cost more than a full CTAS.

Two things would help:

• Doc honesty: say the win is compute-per-row, and it only pays off when that dominates the base+target scan.
• On the pruning lever, a subtlety worth recording: PARTITION BY doesn't actually fit this design. BigQuery partitions on a date/timestamp/int column and prunes a MERGE only with a bounded range predicate on it — the ON T.pk = S.pk equality on an arbitrary key gives neither. True partition pruning needs the event-time-watermark strategy the PR already defers. The lever that does fit merge-by-PK is clustering by the merge key (CLUSTER BY <pk> in the seed CTAS): BigQuery/Snowflake can block-prune the target scan on the equality join, no time column needed. Opt-in and dialect-gated (Postgres has no equivalent), so a follow-up rather than a v0 blocker — but it's the specific "richer strategy later" this design should point at instead of partitioning. dbt's own BigQuery guidance is the reference: MERGE scans the whole destination by default, and clustering on the unique key is their mitigation for the merge case (dbt-bigquery configs).

4. The stable-target-name requirement is an unenforced contract on the orchestrating caller.
Incremental's seed-vs-MERGE decision keys entirely off whether the target table exists, so the source's physical name must be stable across runs. In auto mode that holds — selfAssignTableName (:628) derives it from the persist name. But in orchestrated mode the caller-supplied instructions are used verbatim (instructions = opts.buildInstructions!, :502; the incremental path reads instruction.physicalTableName at :1049), and validateInstructions (:387) has no stability check. If an orchestrating layer assigns a name that varies per version or per content hash, every run misses the existing table and re-seeds — a silent full rebuild that quietly defeats the feature while reporting success. Right now the only guard is the doc's "Stable target name (assumption)" note. Suggest promoting it from assumption to an enforced precondition: in orchestrated mode, reject (or at least loudly warn) when an incremental source's supplied physical name isn't stable, and state it as a hard requirement in the build-instruction API contract rather than a doc footnote.

One more, lower-confidence and rollout-dependent: during a rolling deploy, can a scheduled materialization run land on a pre-#913 replica? If so it treats the source as full-rebuild, never hydrates the CTE, and re-pays the entire per-row cost the feature exists to avoid — correct data, but the expensive full recompute, replacing the accumulated table. Bounded and self-healing if the deploy is effectively atomic for the scheduler; just want to confirm which.