Streamform: Plan identity

A streaming application is persistent: it holds state, and a new version of it must either reuse that state or rebuild it. Deciding which needs two things from every version of the plan: a way to say “this is the same node” across versions, and a way to say “this node stores the same thing”. This document describes both, what moves them, and where they are written down. Comparing two versions is streamform plan; this is the material it compares.

Node identity

Every node in a plan has a stable id of the form <model>::<kind>::<ordinal>:

customer_metrics::source::0
customer_metrics::aggregate::0
customer_metrics::project::0
customer_metrics::sink::0

The ordinal counts operators of that kind within that model, in plan order. Identity is deliberately positional and independent of what the node does, so that changing an aggregate’s key or accumulators reports the same node, changed, rather than one node removed and another added. That is what lets an upgrade say “the aggregate in customer_metrics changed its key” instead of “rebuild”.

What moves an id: renaming the model, or adding a second operator of the same kind before an existing one. What does not: editing whitespace, reordering projection columns, changing a key, adding or removing a WHERE, changing an accumulator.

State fingerprint and digest

A node that keeps state (today, an aggregate) has a state fingerprint: the operator kind, the key columns in order, each accumulator by name and type, and the retention — unbounded, or the model’s declared retention rendered canonically (30d, 90m), see Event time:

aggregate key=[customer_id] accumulators=[sum(amount): decimal(38,9)] retention=unbounded

Every name in it is Streamform’s own. Accumulators are named by Streamform’s rendering of the call (sum(amount), count(*)), never by an alias or by the SQL planner’s internal name, and types use the same vocabulary as sources.yml. The accumulator type is derived by Streamform’s rule from the argument’s declared type: a SUM over decimal(10,2) accumulates decimal(38,2), a COUNT accumulates int64.

The digest is the SHA-256 of that text, 64 lowercase hex characters; text output shows the first 12. Because the text depends only on what the user wrote and on Streamform’s own rules, the digest cannot move when a dependency or the Rust toolchain is upgraded. It moves exactly when the state changes meaning: a different key, a different accumulator, a different type.

Schema fingerprint

Every node also has a schema fingerprint: its output columns by name and canonical type, in order.

[customer_id: int64, total_spend: decimal(38,9)]

Nullability is not part of it; it is inferred by the frontend and declared nowhere in a project. An aggregate’s output columns are named and typed like its accumulators (sum(amount): decimal(38,9)); the alias in the SQL is applied by the projection above it (sum(amount) AS total_spend). A model’s contract with its readers, ref() in another model or an external consumer, is its Sink’s schema fingerprint.

Changelog contract

Every node records how its output changes over time: append (every row is a new fact), upsert(<key>) (a later row for a key supersedes the earlier one), or retract. A model’s Sink records the declared materialization and key as well.

Seeing it

streamform inspect-plan [MODEL] prints the identity of one model, or of every model in dependency order followed by the application digest:

customer_metrics
  materialized=upsert key=[customer_id]
  output=customer_metrics::sink::0

  customer_metrics::source::0  Source
    definition=orders (kafka orders)
    schema=[amount: decimal(38,9), customer_id: int64]  digest=55c5eb27e8b4
    boundedness=unbounded changelog=append
    state=stateless

  customer_metrics::aggregate::0  Aggregate
    definition=key=[customer_id] aggregates=[sum(amount)]
    schema=[customer_id: int64, sum(amount): decimal(38,9)]  digest=322eb3aa5330
    boundedness=unbounded changelog=upsert(customer_id)
    state=keyed key=[customer_id] accumulators=[sum(amount): decimal(38,9)] retention=unbounded
    state digest=4aead023aa01

  customer_metrics::project::0  Project
    definition=customer_id, sum(amount) AS total_spend
    schema=[customer_id: int64, total_spend: decimal(38,9)]  digest=7fac6038b2ce
    boundedness=unbounded changelog=upsert(customer_id)
    state=stateless

  customer_metrics::sink::0  Sink
    definition=upsert kafka customer-metrics
    schema=[customer_id: int64, total_spend: decimal(38,9)]  digest=7fac6038b2ce
    boundedness=unbounded changelog=upsert(customer_id)
    state=stateless

streamform explain shows the same plan operator by operator, with the state digest under the fingerprint of each keyed operator.

The plan manifest

streamform inspect-plan --json prints the plan manifest, a versioned JSON document holding everything above for the whole application. streamform build writes the same document to build/plan.json next to the backend output, so every build carries the identity it was built from. streamform explain --json MODEL prints one model’s slice of it.

{
  "manifest_version": 1,
  "application_digest": "…",
  "models": [
    { "name": "customer_metrics", "materialized": "upsert", "key": ["customer_id"], "output": "customer_metrics::sink::0" }
  ],
  "nodes": [
    {
      "id": "customer_metrics::aggregate::0",
      "model": "customer_metrics",
      "operator": "aggregate",
      "inputs": ["customer_metrics::source::0"],
      "definition": "key=[customer_id] aggregates=[sum(amount)]",
      "schema": [
        { "name": "customer_id", "type": "int64" },
        { "name": "sum(amount)", "type": "decimal(38,9)" }
      ],
      "schema_digest": "…",
      "boundedness": "unbounded",
      "changelog": { "mode": "upsert", "key": ["customer_id"] },
      "state": {
        "kind": "keyed",
        "keys": ["customer_id"],
        "accumulators": [ { "name": "sum(amount)", "type": "decimal(38,9)" } ],
        "retention": "unbounded",
        "digest": "…"
      }
    }
  ]
}

A node that carries a time attribute (a source with event_time, and every node downstream that keeps the column) has a time field naming the column; it is omitted when there is none, and the Source’s definition gains event_time=… watermark=… late=…. A project without event time therefore produces a manifest byte-identical to one written before event time existed.

application_digest is the SHA-256 of the manifest’s compact JSON with that field empty. A reader refuses a manifest from another manifest_version or with fields it does not know, rather than interpreting it. The manifest is derived from the plan and is not turned back into one: nothing executes from it, and streamform plan compares two of them.

Comparing two of them

streamform plan compares the manifest that is running (recorded by streamform apply, or any manifest named with --against) with the one compiled from the project, classifies every change, and says what continuing from the existing state would need. A model that was renamed, or a node whose ordinal shifted, can be matched to its running counterpart with renamed_from and nodes.<kind>.<ordinal>.was in streamform.yml. See the plan diff.