Streamform: The plan diff

A streaming application is persistent. Changing a model and deploying it does not start from nothing: there is state, accumulated under the previous version, and there are rows already written to every sink. streamform plan compares the plan that is running with the plan the project now describes and says, node by node, what the change costs before anything is deployed. This document describes the comparison, the five classifications and the rules behind them, the record of what is running, and the hints that tell the diff a node moved.

What is compared

Both sides are plan manifests (see plan identity): the desired one is compiled from the project on disk, the running one is what streamform apply recorded for the target, or any manifest or record named with --against. Nodes are matched by id. A node on one side only is added or removed; a node on both sides is compared on its definition, schema, boundedness, changelog contract, and state.

streamform plan                       # against the target's applied record
streamform plan --target staging
streamform plan --against build/plan.json
streamform plan --json

The five classifications

Every change is classified with one of five words. They are ordered, and the most severe one wins: a model’s verdict is the most severe of its nodes’, the application’s the most severe of its models’.

Classification	Meaning	Action
`SAFE`	Nothing that affects state or output changed	none
`COMPATIBLE`	Something changed, but no existing state is affected and past output stands	redeploy; existing state continues
`STATE MIGRATION REQUIRED`	Existing state has the right key but not the right shape; it can be transformed mechanically	migrate state before the new plan continues from it
`BACKFILL REQUIRED`	Existing (or new) state can be loaded, but its contents would not match a from-scratch run under the new plan	replay history through the new plan
`STATE INCOMPATIBLE`	Existing state cannot be reused at all	rebuild

STATE INCOMPATIBLE is the most severe because a rebuild implies a backfill; BACKFILL REQUIRED is more severe than a migration because a migration keeps the state’s contents and a backfill recomputes them.

The rules

For a keyed node (an aggregate) present on both sides:

What changed	Classification	Why
nothing	`SAFE`
the key columns, or their order	`STATE INCOMPATIBLE`	state is partitioned and addressed by the key; there is no mapping from old keys to new
the type of an existing accumulator	`STATE INCOMPATIBLE`	the stored value has a different encoding
an accumulator added	`BACKFILL REQUIRED`	the new accumulator starts at its initial value for every existing key; a from-scratch run would have accumulated history into it
an accumulator removed	`STATE MIGRATION REQUIRED`	drop it; every remaining value is exactly what a from-scratch run would hold
the retention	`STATE MIGRATION REQUIRED`	timers change, values do not — reachable once a model declares `retention` (event time)
its input	`BACKFILL REQUIRED`	existing state was accumulated from the previous input
the clock of a source upstream (`event_time`, `watermark`, `late_events`)	`BACKFILL REQUIRED`	the clock decides which events are late, so it decides which events reached the state; see event time

For a stateless node (source, filter, projection, sink) present on both sides, any change is COMPATIBLE: there is no state to migrate. Two exceptions on a sink, because a sink topic is state the reader holds:

What changed on a sink	Classification	Why
the declared key, or `materialized`	`BACKFILL REQUIRED`	rows already in an upsert topic are keyed the old way; the compacted topic becomes a mixture unless it is rebuilt
the connector or target	`COMPATIBLE`	output moves; the old topic is left as it is, and the message says so
the schema	`COMPATIBLE`	readers see new columns; rows already written keep the old shape, and the message says so

Added and removed nodes: a stateless node added or removed is COMPATIBLE; a keyed node added is BACKFILL REQUIRED (its state starts empty, and history is in it only if the source replays it); a keyed node removed is COMPATIBLE, and the message says its state is discarded when the old job stops. A whole model added or removed is classified as its nodes.

Propagation

A keyed node whose own state is unchanged, but which has an upstream node whose definition changed, is raised to BACKFILL REQUIRED. Its state was accumulated under the previous definition of its input: a WHERE that used to drop refunds and no longer does, a source that now reads a different topic. Loading that state under the new plan continues from totals a from-scratch run would never produce. Propagation follows inputs all the way up and across model boundaries: a changed filter in clean_orders raises the aggregate in customer_metrics, the model that reads it.

This rule is deliberately conservative. Streamform does not yet track which columns a node reads, so a projection that adds a column the aggregate never uses triggers it too. A tool whose job is to prevent silently wrong totals errs towards flagging; column-level lineage will make the rule precise later.

Nothing running

When nothing is recorded as applied to the target and no --against is given, every model is new and the verdict is SAFE: there is no existing state to contradict, so the application simply starts.

Reading the output

The example project, recorded as v0.3.0 wrote it, after changing GROUP BY customer_id to GROUP BY customer_id, amount (and the model’s key with it):

plan: customer-metrics against target `local`
  running: applied 2026-08-27T10:12:03Z by streamform 0.4.0, job 9d1f2c4b7a3e, application digest 7b0d9542cf02
  desired: application digest d1fbf1d3c6ab

customer_metrics  changed
  customer_metrics::aggregate::0  Aggregate  changed
    definition: key=[customer_id] aggregates=[sum(amount)] → key=[customer_id, amount] aggregates=[sum(amount)]
    schema: [customer_id: int64, sum(amount): decimal(38,9)] → [customer_id: int64, amount: decimal(38,9), sum(amount): decimal(38,9)]
    changelog: upsert(customer_id) → upsert(customer_id, amount)
    key: [customer_id] → [customer_id, amount]
    state digest: 4aead023aa01 → d982cde38013
    STATE INCOMPATIBLE: the grouping key changed; existing state cannot be reused
  customer_metrics::project::0  Project  changed
    definition: customer_id, sum(amount) AS total_spend → customer_id, amount, sum(amount) AS total_spend
    schema: [customer_id: int64, total_spend: decimal(38,9)] → [customer_id: int64, amount: decimal(38,9), total_spend: decimal(38,9)]
    changelog: upsert(customer_id) → upsert(customer_id, amount)
    COMPATIBLE: stateless; no stored value is affected
  customer_metrics::sink::0  Sink  changed
    schema: [customer_id: int64, total_spend: decimal(38,9)] → [customer_id: int64, amount: decimal(38,9), total_spend: decimal(38,9)]
    changelog: upsert(customer_id) → upsert(customer_id, amount)
    key: [customer_id] → [customer_id, amount]
    BACKFILL REQUIRED: rows already written to `customer-metrics` are keyed the old way; rebuild the sink
  STATE INCOMPATIBLE: rebuild required

verdict: STATE INCOMPATIBLE (1 model to rebuild)
note: the flink backend does not carry state across a changed query; on this target the action is a rebuild, and the classification tells you what a rebuild costs

The header says what is being compared: when the running plan was applied, by which Streamform, the job it started, and the application digests of both sides. Each changed node lists the fields that differ with their old and new values, then its classification and the reason. Unchanged nodes and models are not listed beyond SAFE. The verdict line counts models by what they need.

Exit codes: 0 when the verdict is SAFE or COMPATIBLE, so the application can continue from its existing state; 1 for any other verdict; 2 when the project does not compile or the running side cannot be read. CI can gate on it.

--json prints the same as a document (diff_version: 1): the target, both sides’ digests, the verdict, every model with its verdict and changed nodes, each node’s change (added, removed, changed), classification, reason, and the fields that differ, and any warnings.

What the classification is, and is not

The classification is semantic: it says what existing state would need in order to continue under the new plan. Whether a backend can do that is the backend’s business. Flink SQL restores state only for a query whose topology and serializers are unchanged, and exposes no operator ids for Streamform to pin, so today on Flink every change that is not SAFE is a rebuild in practice. plan says so in one line whenever the target’s backend is Flink and the verdict is above SAFE. Nothing performs a migration or a backfill yet; the diff tells you what is required, and the native runtime is where Streamform will own the state and execute it.

streamform apply runs the same comparison against the target’s record before applying and prints the verdict, the whole diff when it cannot continue, but it does not refuse. It cannot stop, savepoint, or restore a job yet, so on Flink every apply starts a new job with empty state whatever the verdict says; a gate that had to be bypassed for every real upgrade would only teach people to bypass it. Until job lifecycle exists, plan’s exit code is the gate.

The applied record

After the last statement is accepted, streamform apply writes what it applied to .streamform/applied/<target>.json:

{
  "record_version": 1,
  "target": "local",
  "applied_at": "2026-08-27T10:12:03Z",
  "streamform_version": "0.4.0",
  "backend": "flink",
  "jobs": ["9d1f2c4b7a3e5f6081c2d3e4f5a6b7c8"],
  "manifest": { "manifest_version": 1, "application_digest": "7b0d…", "models": [], "nodes": [] }
}

The manifest is embedded verbatim, so a record is also a manifest and --against accepts either, told apart by the top-level version field. The record is written atomically, so a rejected apply leaves the previous one in place. .streamform/ is the project’s record directory, the same role .terraform/ plays; whether a team commits its records or keeps them in a CI cache is the team’s call, and the file contains nothing secret. A record can drift from reality (a job cancelled from the Flink UI, an apply from a machine whose record was never shared); the header prints when and by whom it was applied so that drift is visible.

A reader refuses a record from another record_version, a manifest from another manifest_version, unknown fields, and a file that is neither a record nor a manifest, rather than interpreting it.

Identity hints

Ids are positional: <model>::<kind>::<ordinal>. Two edits move them: renaming a model, and inserting a second operator of the same kind before an existing one. Without help the diff reports the renamed model as removed and a new one added, and the new aggregate as BACKFILL REQUIRED. Two hints in streamform.yml tell the diff which running node a desired node was:

models:
  customer_totals:
    materialized: upsert
    key: [customer_id]
    renamed_from: customer_metrics          # every customer_totals::<kind>::<n> was customer_metrics::<kind>::<n>
    nodes:
      aggregate.0:
        was: customer_metrics::aggregate::1  # this node in particular was that one

renamed_from matches every node of the model by swapping the model prefix; nodes.<kind>.<ordinal>.was matches one node and takes precedence. Ids in the desired plan are not changed; the hints only choose the counterpart to compare with. After apply, the record holds the new ids and the hints do nothing: plan warns about a hint that names no running node, which is the cue to delete it. streamform check rejects a hint whose node reference is not <kind>.<ordinal>, whose was is not a node id, a renamed_from naming the model itself or a model that still exists, two hints on one running node, and a hint for a node the model’s plan does not have.

A hint makes the diff say “same node, changed” instead of “removed, added”. It does not move state; whether the backend can carry state across a rename is the separate fact described above.

Keyboard shortcuts

Streamform