Upgrading a running application
A running aggregation remembers every customer’s total. Change the model and deploy, and the question is whether that memory still means anything. streamform plan answers it before anything is deployed: it compares the plan that is running with the plan the project now describes and classifies every change. This guide walks through the workflow on the example project.
Apply, and let Streamform remember what ran
streamform apply
plan: nothing is recorded as applied to target `local`; the application starts fresh
applying customer-metrics to target `local` (http://localhost:8083)
ok: source orders
ok: model customer_metrics
ok: sink customer_metrics
ok: statement set → job 9d1f2c4b7a3e5f6081c2d3e4f5a6b7c8
ok: applied
ok: applied plan recorded in .streamform/applied/local.json
The last line is new. apply now writes the plan it applied, with the time, the Streamform version, and the job id, to .streamform/applied/<target>.json. That file is the running plan as far as streamform plan is concerned. Commit it or keep it in your CI cache, as you prefer; it holds nothing secret.
Without a Flink to apply to, streamform build --backend flink writes build/plan.json, and plan --against build/plan.json compares with that instead. Everything below works the same way.
Ask before you change anything
Edit nothing and run:
streamform plan
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 7b0d9542cf02
customer_metrics SAFE
verdict: SAFE
Exit code 0. Now the edits. Add an accumulator:
SELECT customer_id, SUM(amount) AS total_spend, COUNT(*) AS orders
FROM source('orders')
GROUP BY customer_id
customer_metrics changed
customer_metrics::aggregate::0 Aggregate changed
definition: key=[customer_id] aggregates=[sum(amount)] → key=[customer_id] aggregates=[sum(amount), count(*)]
schema: [customer_id: int64, sum(amount): decimal(38,9)] → [customer_id: int64, sum(amount): decimal(38,9), count(*): int64]
accumulators: [sum(amount): decimal(38,9)] → [sum(amount): decimal(38,9), count(*): int64]
state digest: 4aead023aa01 → a5771fc8a5b2
BACKFILL REQUIRED: accumulator `count(*)` added; it starts empty for every existing key, replay history to fill it
customer_metrics::project::0 Project changed
definition: customer_id, sum(amount) AS total_spend → customer_id, sum(amount) AS total_spend, count(*) AS orders
schema: [customer_id: int64, total_spend: decimal(38,9)] → [customer_id: int64, total_spend: decimal(38,9), orders: int64]
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, total_spend: decimal(38,9), orders: int64]
COMPATIBLE: readers of `customer-metrics` see the new columns; rows already written keep the old shape
BACKFILL REQUIRED: replay history through the new plan
verdict: BACKFILL REQUIRED (1 model to backfill)
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
Exit code 1. The aggregate is the same node (customer_metrics::aggregate::0, same key) with one more accumulator. Its existing state could be loaded, but every customer’s count(*) would start at zero while their sum(amount) carries history: not what a fresh run would produce. Hence BACKFILL REQUIRED. The projection and the sink change too, but they hold no state, so they are COMPATIBLE, and the sink’s message says what readers will see.
Now change the grouping key instead (GROUP BY customer_id, amount, and key: [customer_id, amount] in streamform.yml):
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::sink::0 Sink changed
...
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)
Same node, different partitioning: nothing maps old keys to new, so the state cannot be reused at all, and the sink’s compacted topic is keyed the old way as well.
The change that hides upstream
The diff is most useful where the edit is far from the state. In examples/order-pipeline, customer_metrics aggregates clean_orders, a filter. Tighten the filter (WHERE customer_id IS NOT NULL AND amount > 0) and ask:
clean_orders changed
clean_orders::filter::0 Filter changed
definition: (customer_id IS NOT NULL) → ((customer_id IS NOT NULL) AND (amount > 0))
COMPATIBLE: stateless; no stored value is affected
COMPATIBLE: redeploy; existing state continues
customer_metrics changed
customer_metrics::aggregate::0 Aggregate changed
BACKFILL REQUIRED: input definition changed upstream at `clean_orders::filter::0`; existing state was accumulated under the previous definition
BACKFILL REQUIRED: replay history through the new plan
big_customers SAFE
verdict: BACKFILL REQUIRED (1 model to backfill, 1 model changed)
Nothing about the aggregate itself changed; its state digest is the same. But every total in it was accumulated from orders the new filter would have dropped, so continuing from that state gives numbers a fresh run would never produce. The diff names the upstream node. big_customers, a stateless filter over customer_metrics, is SAFE: it holds nothing.
This rule is conservative on purpose. Streamform does not yet know which columns an aggregate reads, so adding an unused column upstream triggers it too. The plan diff reference has the full rule table.
Renames and restructures
Ids are positional (customer_metrics::aggregate::0), so renaming the model to customer_totals makes the diff report one model removed and another added, with its aggregate BACKFILL REQUIRED. Tell it what happened:
models:
customer_totals:
materialized: upsert
key: [customer_id]
renamed_from: customer_metrics
customer_totals SAFE
verdict: SAFE
The same works for one node, when a restructure shifts an ordinal: nodes: { aggregate.0: { was: customer_metrics::aggregate::1 } }. Hints only choose which running node to compare with; they do not change ids. Once the rename is applied, the record holds the new names and plan warns that the hint is unused: delete it.
What apply does with the verdict
streamform apply runs the same comparison first and prints the verdict; when the application cannot continue from its state it prints the whole diff, then applies anyway, saying so. It cannot stop or restore a Flink job yet, so today every apply starts a new job with empty state whatever the verdict, and refusing would only teach people to bypass the refusal. Use plan’s exit code as the gate in CI:
streamform plan --target production || exit 1
What the verdict does not do
The classification is about the plan: what existing state would need in order to continue. Flink SQL restores state only for an unchanged query, so on Flink every change above SAFE is a rebuild in practice, and plan says so in its last line. Nothing performs a migration or a backfill yet. The plan diff reference is exact about the rules, the record, and the hints; the plan identity reference is exact about what is compared.