Streamform

Streamform is a workflow for building, testing, running, and upgrading stateful streaming applications from declarative SQL models. You write SQL, declare where the input comes from and how the result should be materialized, and Streamform compiles that into a continuous dataflow it can test in-process and run on Apache Flink.

It is one Rust binary, streamform, and a project is a directory of plain files. There is no JVM on your machine, no service to run, and nothing generated that you cannot read.

The one idea everything follows from

Take the most ordinary aggregation there is:

SELECT customer_id, SUM(amount) AS total_spend
FROM source('orders')
GROUP BY customer_id

Over a stream, its input is append-only: an order for customer 42 of 10, then another of 18. But its output is not a growing list. It is a relation that changes:

insert   customer_id=42  total_spend=10
update   customer_id=42  total_spend=28

A SQL statement says nothing about that. It does not say whether orders ends, that the result is keyed by customer_id, that computing it needs the runtime to remember every customer’s total, or what should happen to that memory when you change the query and deploy again. Today that knowledge lives in the person who knows Flink.

Streamform makes it part of the program. A model is the SQL plus three declarations the SQL cannot make:

From those, Streamform derives the rest: whether each step is bounded, what changelog it emits, what state it holds, and a stable identity for that state so a later version can be compared against it.

The workflow

streamform check          validate the project against its compiled plan
streamform explain        read a model as a streaming plan
streamform test           run fixtures through the simulator, assert the changelog
streamform build          generate the application as Flink SQL
streamform apply          submit it through the Flink SQL Gateway
streamform inspect-plan   see the identity and digests an upgrade compares
streamform plan           compare the running plan with the desired one and classify every change

Every command compiles the project to the same plan and works from it, so what you test is what you deploy and what you inspect.

Where to start

• Install the binary and run the first project in about a minute.
• Read the concepts if you want to know what explain is telling you.
• The reference is exact about every command, file, and exit code.
• Why Streamform holds the founding documents, for the reasoning behind the design.

Status

Streamform is early and says so. Version 0.4 ships single- and multi-model applications with append and upsert changelogs, the deterministic simulator, the Flink backend, a stable plan identity for every node, and streamform plan, which classifies every change before it is deployed. Built since, and shipping as version 0.5: event time, with watermarks, late events that a fixture must acknowledge, and retention. Not yet: windows, joins, job lifecycle, dbt and Python frontends. The roadmap is ordered by dependency and evidence, not by date, and each capability lands whole (project format, simulator, Flink, plan diff, documentation) before it is called shipped. The project format is versioned and unknown fields are rejected, so an older build never silently misreads a newer project.

Install

Streamform is a single binary, free to use locally and in production with no account (what you may do with it). Nothing else is needed to check, explain, and test a project; Apache Flink is needed only when you build for it and apply.

Prebuilt binaries

Each release on the releases page ships tarballs for macOS (Apple silicon and Intel) and Linux x86_64, plus a checksums.txt. Verify the download against it, unpack, and put streamform on your PATH. On Apple silicon, for the latest release:

curl -fsSL https://github.com/glyf-data/streamform-releases/releases/latest/download/streamform-aarch64-apple-darwin.tar.gz | tar xz
sudo mv streamform /usr/local/bin/
streamform --version

Replace the target with x86_64-apple-darwin or x86_64-unknown-linux-gnu as needed; pin a version by replacing latest/download with download/<tag>, for example download/v0.5.0.

The example projects

The projects this site walks through are published beside the binary, together with a local Kafka and Flink environment for streamform apply:

curl -fsSL https://github.com/glyf-data/streamform-releases/releases/latest/download/streamform-examples.tar.gz | tar xz
cd streamform-examples

checksums.txt covers this archive too.

Homebrew, signatures, containers

Planned, and not available yet: brew install, signed release artifacts, a container image, and builds for Windows and aarch64 Linux. Until they exist, the tarball and its SHA-256 checksum are the install path.

What else you might want

Working from the repository (maintainers)

The source repository is private. It uses mise to pin the toolchain and run tasks:

git clone https://github.com/glyf-data/streamform
cd streamform
mise run setup                 # toolchain, dependencies, git hooks
mise run cli -- --version      # run the CLI from source
mise run check                 # everything CI runs

Without mise, install Rust 1.97 with rustfmt and clippy and use cargo directly.

What you may do with it

Streamform is three things, and each has one rule.

Why it is drawn this way

Everything public is a recipe, a schema, or a client of the binary — never a piece of it. The formats are published so that your tools, your CI, and the open testing library can read what Streamform writes without needing its source. The engine stays closed because it is the work; it stays free because a tool for developing streaming applications is only useful if every developer on the team can run it.

Where things are

• Downloads, checksums, the example projects, and the changelog (signatures are planned): glyf-data/streamform-releases. Issues and questions go there too.
• The formats: glyf-data/streamform-spec, mirrored from the binary’s own source on every release, starting with v0.5. streamform schema <document> prints the same schemas from the binary you have.
• The testing library, the runner image, and the Action: planned, with no repository yet.

Your first project

Every release publishes the example projects beside the binary. customer-metrics is the smallest project that shows the whole idea: one source, one model, one fixture. Everything below is real output from that project.

curl -fsSL https://github.com/glyf-data/streamform-releases/releases/latest/download/streamform-examples.tar.gz | tar xz
cd streamform-examples/customer-metrics

Three files

The source: a Kafka topic, so the input never ends. The schema names the columns and their types.

# sources.yml
sources:
  orders:
    connector: kafka
    topic: orders
    format: json
    schema:
      customer_id: int64
      amount: decimal

The model: plain SQL. source('orders') reads the source declared above.

-- models/customer_metrics.sql
SELECT
    customer_id,
    SUM(amount) AS total_spend
FROM source('orders')
GROUP BY customer_id

The project file: the model’s result is keyed by customer_id and materialized as an upsert. Streamform checks that this agrees with what the SQL produces; a GROUP BY without upsert is an error, not a guess.

# streamform.yml
name: customer-metrics
version: 1

models:
  customer_metrics:
    materialized: upsert
    key:
      - customer_id

Check it

streamform check

customer-metrics (schema version 1)

models
  customer_metrics         upsert(customer_id)      models/customer_metrics.sql

sources
  orders                   kafka/json   topic=orders             2 columns

fixtures
  customer_metrics         tests/customer_metrics.yml 2 given, 2 expect

ok: 1 model, 1 source, 1 fixture, 0 errors, 0 warnings

check compiles every model to its streaming plan and reports anything that does not plan: an unsupported construct, a ref() to a model that does not exist, a declared key that disagrees with the GROUP BY, a fixture that names a column the model does not output.

Read what the SQL means

streamform explain customer_metrics

customer_metrics

Source
  name=orders
  connector=kafka
  time=processing
  boundedness=Unbounded
  changelog=Append
  state=Stateless
     ↓
Aggregate
  key=customer_id
  aggregates=sum(amount)
  boundedness=Unbounded
  changelog=Upsert(customer_id)
  state=Keyed
  fingerprint=aggregate key=[customer_id] accumulators=[sum(amount): decimal(38,9)] retention=unbounded
  digest=4aead023aa01
     ↓
Project
  columns=customer_id, sum(amount) AS total_spend
  boundedness=Unbounded
  changelog=Upsert(customer_id)
  state=Stateless
     ↓
Sink
  model=customer_metrics
  materialized=upsert
  boundedness=Unbounded
  changelog=Upsert(customer_id)
  state=Stateless

Read it top to bottom: the source never ends, grouping it needs keyed state (one accumulator per customer, a decimal(38,9) running total), and from the aggregate onward the result is an updating stream keyed by customer_id. The concepts section explains each line.

Test it

The fixture gives the model two events and states the exact changelog it must emit.

# tests/customer_metrics.yml
model: customer_metrics

given:
  - customer_id: 42
    amount: 10
  - customer_id: 42
    amount: 18

expect:
  - op: insert
    customer_id: 42
    total_spend: 10
  - op: update
    customer_id: 42
    total_spend: 28

streamform test

customer_metrics

PASS

2 events processed
2 changelog records emitted

ok: 1 fixture passed

That ran through the in-process simulator: no Kafka, no cluster, no wall clock, and the same result every time. Change the second expectation to total_spend: 27 and the failure shows expected and actual side by side, with the first difference named:

  #  expected                              actual
  1  insert customer_id=42 total_spend=10  insert customer_id=42 total_spend=10
  2  update customer_id=42 total_spend=27  update customer_id=42 total_spend=28
     ^ total_spend: expected 27, got 28

Next

• Project layout for what each file may contain.
• Testing with fixtures for multi-model fixtures and the exact comparison rules.
• Running on Apache Flink to build and apply this same project.

Project layout

A project is a directory. Every file in it is a contract Streamform keeps: the formats are versioned, and a change to any of them is a documented change.

streamform.yml     project name, format version, per-model materialization, targets
sources.yml        the external inputs and their schemas
models/*.sql       one SQL model per file; the file stem is the model name
tests/*.yml        fixtures, one per file
build/             output of `streamform build`; not source, keep it out of version control

streamform.yml

Names the project, pins the format version (1), and configures models and targets. A model exists because models/<name>.sql exists; the entry under models: only configures it. A model with no entry is append with no key.

name: order-pipeline
version: 1

models:
  customer_metrics:
    materialized: upsert
    key: [customer_id]
  big_customers:
    materialized: upsert
    key: [customer_id]
    sink:
      connector: kafka
      topic: big-customers
      format: json

targets:
  local:
    backend: flink
    flink:
      gateway: http://localhost:8083
    kafka:
      bootstrap_servers: kafka:9092

• materialized is append (the default) or upsert, and must agree with what the SQL produces: a GROUP BY produces an updating result and requires upsert with a key equal to the grouping columns.
• sink adds an external destination for the model’s changelog. Every model ends in a Sink in its plan whether or not it declares one; the Sink is what fixtures assert on and what ref() in another model reads. A sink: block is what makes it leave the application.
• targets name the places an application is built for and applied to. See Running on Apache Flink.

sources.yml

Each source has a connector, a location, a format, and a schema. The connector decides whether the input ends: kafka is unbounded, file is bounded. That is derived, never guessed from the SQL.

sources:
  orders:
    connector: kafka
    topic: orders
    format: json
    schema:
      customer_id: int64
      amount: decimal(12,2)

Types: int32, int64, float64, decimal (which is decimal(38,9)), decimal(precision,scale), string, boolean, timestamp. The declared scale of a decimal is what state keeps: a SUM over decimal(10,2) accumulates decimal(38,2).

models/*.sql

One SELECT per file. Read a source with source('<name>') and another model with ref('<model>'); a plain table name is not valid, because the dependency graph would not know about it.

The supported surface today: column references, literals, comparison and arithmetic, AND/OR/NOT, IS [NOT] NULL, CAST, WHERE, GROUP BY over columns, HAVING, and sum, count, min, max. Anything else fails at check naming the construct, rather than reaching a backend that would run it differently.

tests/*.yml

A fixture names a model, gives it events, and states the exact changelog it must emit. Testing with fixtures walks through it; the project format reference is exact about every field and every comparison rule.

The streaming plan

Everything Streamform does goes through one representation: the streaming plan. check validates against it, explain prints it, test executes it in the simulator, build generates Flink SQL from it, and inspect-plan describes its identity. Because every command reads the same plan, what you test is what you deploy.

From SQL to a plan

A model’s SQL is parsed and planned by DataFusion, then immediately lowered into Streamform’s own intermediate representation and DataFusion is not consulted again. The IR is deliberately small: five operators and a seven-variant expression tree covering exactly what Streamform supports today. SQL that does not fit fails at check, naming the construct, instead of reaching a backend that would run it differently from the simulator.

What every node knows

Each node carries three properties that are first-class in Streamform rather than details of an engine:

• Boundedness: whether the node’s output ever ends. Derived from the source’s connector and carried through. See Boundedness.
• Changelog mode: how the output changes over time: Append, Upsert(key), or Retract. See Changelogs.
• State requirement: Stateless, or Keyed with a fingerprint of exactly what is stored. See State.

explain prints them under each operator:

Aggregate
  key=customer_id
  aggregates=sum(amount)
  boundedness=Unbounded
  changelog=Upsert(customer_id)
  state=Keyed
  fingerprint=aggregate key=[customer_id] accumulators=[sum(amount): decimal(38,9)] retention=unbounded
  digest=4aead023aa01

One plan per application

A project with several models compiles to one plan. Models appear in dependency order, each as a chain ending in its Sink, and ref('m') is an edge from a model’s first operator to m’s Sink. A changelog mode crosses that edge: a model reading an upsert model receives updates, not fresh facts, and its own operators are planned accordingly. See Applications of many models.

Nothing engine-specific inside

The plan contains nothing that belongs to Flink, and the Flink backend is generated from it alone. That is a design rule with a test behind it: if a feature ever needs an engine concept inside the plan, the plan is what gets revisited, not the backend. It is also what makes the simulator an honest oracle for Flink, and what will let a native runtime consume the same plan later.

Boundedness

Does the input end? A file does. A Kafka topic does not. The same SQL means different things over each, so Streamform records the answer on every node of the plan and refuses SQL whose meaning depends on an ending that never comes.

Where it comes from

Boundedness is derived from the source’s connector, never guessed from the SQL:

Every operator downstream carries its input’s boundedness forward. explain shows it on each node as boundedness=Unbounded or boundedness=Bounded.

What it changes

A global ORDER BY needs the whole input to produce a final ordering. Over a bounded file that is a sort; over an unbounded topic it can never finish, so Streamform refuses it at check:

error: could not compile model `sorted`: model `sorted`: ORDER BY cannot produce a final
       ordering over an unbounded input; remove it, or sort downstream of a bounded materialization

The same model over a file source is accepted. This is the first and simplest of the semantics a SQL statement alone cannot express, and it is why sources are declared rather than inferred.

Later

Windows, watermarks, and event time (roadmap phases 7 and 8) are the next set of meanings that depend on time and on the input never ending. They arrive as first-class plan properties in the same way.

Changelogs

Streaming is about change. A model’s output is not a table you read once; it is a sequence of changes to a relation, and what kind of changes it emits is a property of the model that Streamform tracks on every node.

Three modes

A GROUP BY is where the mode changes. Its input is facts; its output is a relation that changes as facts arrive, keyed by the grouping columns. A global aggregate (no GROUP BY) is Upsert with an empty key: one row that keeps changing.

Three operations

The changelog a model emits is a sequence of records, each with an operation:

• insert: a row appears.
• update: a row identified by its key takes new values. Fixtures assert the after-image, the values after the change.
• delete: a row leaves.

The first order for customer 42 is an insert of their total; the second is an update. A filter over an updating model turns a value crossing the threshold into an insert when it enters and a delete when it leaves:

-- big_customers.sql
SELECT customer_id, total_spend
FROM ref('customer_metrics')
WHERE total_spend > 100

insert   customer_id=42  total_spend=130
update   customer_id=42  total_spend=140
delete   customer_id=42  total_spend=140      -- a refund took the total under 100

Declared and derived must agree

streamform.yml declares how a model is materialized; the plan derives what the SQL produces; check requires the two to agree.

• A GROUP BY model must declare materialized: upsert with key equal to its grouping columns. A missing key, a different key, or a key on a global aggregate is an error that names the disagreement.
• An append model over an updating input is an error: the updates have nowhere to go.
• upsert over an append stream is allowed when the model declares a key; the key then defines row identity for the sink.

This is deliberate. The materialization is the contract a downstream consumer relies on, and a contract that the compiler can check is worth more than one that is documented.

Where the changelog goes

The Sink at the end of every model carries the model’s changelog. In the simulator, fixtures assert it record by record. On Flink, an upsert model with a Kafka sink becomes an upsert-kafka table with a primary key, so an update is a new value for the key and a delete is a tombstone. Another model reading through ref() receives the same changelog inside the job, and is planned with that mode as its input.

Not yet

Aggregating over an updating input (a GROUP BY over a ref() to an upsert model) is refused today: folding an update into an accumulator means retracting the value it replaces, which is the Retract mode nothing produces yet. It arrives with the roadmap’s retraction work.

State

GROUP BY customer_id asks the runtime to remember something: every customer’s running values. That memory is part of the program, not an implementation detail, and Streamform describes it precisely on the node that holds it.

What a node stores

An Aggregate node keeps, per key, one accumulator per aggregate call. explain shows the description:

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

• Key: the grouping columns, in order. State is partitioned by it.
• Accumulators: one per call, named by Streamform’s rendering of the call (sum(amount), count(*)) and typed by what is stored.
• Retention: how long a key’s state is kept after its last event, in event time. unbounded unless the model declares retention (see Time); expiry is silent, and the key’s next event starts over.

Every other operator today is Stateless.

Accumulator types are Streamform’s rule

The stored type is derived from the argument’s declared type, by a rule Streamform owns:

The simulator stores exactly these types and a test asserts it, so state inference and execution cannot drift. The scale is yours: declare amount: decimal(12,2) and the total is kept to two places.

Why the description is so careful

Two versions of an application must be able to compare what they store, to decide whether the running version’s state can be kept. That comparison is only honest if the description depends on what you wrote and on Streamform’s rules alone, never on how the SQL planner happened to name or widen a column in this release. So the fingerprint uses Streamform’s own names for accumulators and its own vocabulary for types, and its digest is a SHA-256 of that text. See Identity.

The simulator’s state is the same state

streamform test executes the plan with an in-memory keyed store shaped exactly like the fingerprint says: a map from key to accumulators of the recorded types. Nothing is approximated, which is what lets the simulator be the oracle the Flink backend is checked against.

Identity

A streaming application is persistent. Change the SQL and deploy again, and the new version either reuses the state the old one built or rebuilds it from the beginning of the input. Deciding which needs two things from every version of the plan: a way to say “this is the same node”, and a way to say “it stores the same thing”. Streamform gives every node both.

Node ids

Every node has a stable id: <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 the model. Identity is positional on purpose, and independent of what the node does: change the aggregate’s key or its accumulators and it is still customer_metrics::aggregate::0, the same node, changed. That is what lets an upgrade say “the aggregate changed its key” rather than “one node removed, one node added, rebuild”.

What moves an id: renaming the model, or inserting a second operator of the same kind before an existing one. What does not: whitespace, column order, a changed key, an added WHERE, a changed accumulator.

Digests

Every stateful node has a state digest: the SHA-256 of its fingerprint text (aggregate key=[customer_id] accumulators=[sum(amount): decimal(38,9)] retention=unbounded). Every node has a schema digest: the SHA-256 of its output columns by name and canonical type, in order. Text output shows the first twelve characters; the full 64 are what is compared.

Both are computed from Streamform’s own renderings, never from a dependency’s output, so upgrading the compiler’s dependencies or the Rust toolchain cannot move a digest. A digest moves exactly when meaning moves. The digest of the customer_metrics example, 4aead023aa01…, is pinned by a test; changing it is a deliberate act.

The manifest

streamform inspect-plan --json prints all of it for the whole application as a versioned JSON document, the plan manifest, and streamform build writes the same document to build/plan.json. A reader refuses a manifest from another format version or with fields it does not know. See Preparing for upgrades for the workflow and the plan identity reference for the document.

What this is for

streamform plan compares the running manifest with the desired one and classifies every change before anything is deployed: SAFE, COMPATIBLE, STATE MIGRATION REQUIRED, BACKFILL REQUIRED, STATE INCOMPATIBLE. Identity is what lets it say “the aggregate in customer_metrics changed its key” instead of “one node removed, one added”, and it exists early on purpose: identity that is bolted on after users have running state cannot be changed without invalidating that state. See Upgrading a running application and the plan diff reference.

Time

A batch query has no clock: it reads everything, computes, and finishes. A streaming application runs for months, and when an event happened is different from when it arrived. Streamform names both and lets you choose which one a model runs on.

Processing time is the default

A source without event_time runs on processing time: events are processed in arrival order, and a timestamp column is just a column. explain says so on the Source: time=processing. Nothing is late, nothing expires, and a fixture’s given is simply the order of delivery.

Event time gives the plan a clock

Declare event_time: ordered_at on a source and three things follow:

• The time attribute. Every model downstream carries ordered_at as its time attribute for as long as it keeps the column — through a filter, through a projection that selects it (under its alias), and not through an aggregate, which groups it away. explain prints time attribute=… on each node that has one.
• The watermark. The source’s watermark is the largest event time seen so far minus the declared watermark delay, and it never moves backwards. It is the plan’s statement of “everything before this instant has arrived”. watermark: 0s says events are in order; watermark: 5m allows five minutes of disorder.
• Lateness. An event behind the watermark when it is delivered is late. late_events: drop (the default) discards and counts it; keep delivers it as if on time. Dropping is a choice a fixture must acknowledge, so no test loses an event silently.

Retention lets state end

retention: 30d on a model says a key’s state is dropped once the watermark reaches its last event plus thirty days — silently, with no retraction, so the next event for the key starts its aggregate over and an upsert sink replaces the stale row rather than deleting it. Retention is enforced in event time, which is why it needs an event-time source upstream. It fills the retention= of the state fingerprint, so declaring it changes the model’s state digest, and streamform plan classifies the change as STATE MIGRATION REQUIRED.

The simulator is the definition

One watermark per source; late detection at the moment of delivery; expiry after every advancement; no wall clock anywhere. Fixtures can advance the watermark explicitly with - watermark: <timestamp> and must list dropped events under late:. Flink gets the same semantics where Flink SQL can express them — a WATERMARK FOR clause, a late-dropping view, a STATE_TTL hint — and build warns where it can only approximate: retention on Flink is wall-clock time, which is wrong under replay.

The walkthrough with three runnable fixtures is Event time.

Testing with fixtures

A fixture delivers events to a model and asserts the exact changelog the model emits. streamform test runs every fixture under tests/ through the deterministic simulator, in-process, with no services.

Anatomy of a fixture

model: big_customers

given:
  - customer_id: 42
    amount: 60
  - customer_id: 42
    amount: 70
  - customer_id: 42
    amount: 10
  - customer_id: 42
    amount: -50

expect:
  - op: insert
    customer_id: 42
    total_spend: 130
  - op: update
    customer_id: 42
    total_spend: 140
  - op: delete
    customer_id: 42
    total_spend: 140

model names the model under test.

given lists input events, delivered in order to the source at the top of the model’s chain. big_customers reads ref('customer_metrics'), which reads ref('clean_orders'), which reads source('orders'); the events are orders rows, and the changelog asserted is big_customers’ own, after every model in between has run. Columns the source declares but an event omits are null; a column the source does not declare is an error.

expect is the changelog, record by record, in order. The comparison is exact: the same number of records, and record i must have the op and every output column value of expect[i]. An update is asserted by its after-image. Each record names exactly the model’s output columns, and streamform check reports a misspelled or missing column before anything runs.

Values are typed against the schema: an integer 28 matches a decimal column holding 28.000000000; 0.1234 against decimal(38,2) is an error, not a rounding; null matches only null.

What the fixture above proves

Customer 42’s total climbs to 130 (they enter the filter: insert), then 140 (update), then a refund takes it to 90 and they leave (delete). That third record is the reason changelogs are first-class: a WHERE over an updating input must emit a delete when a row stops matching, and a consumer of the big-customers topic depends on receiving it.

Running

streamform test                     # every fixture
streamform test --model big_customers

Each fixture prints its model, PASS or FAIL, the events processed, and the records emitted. A failure adds a side-by-side listing with the first difference named:

customer_metrics

FAIL
  --> tests/customer_metrics.yml

2 events processed
2 changelog records emitted

  #  expected                              actual
  1  insert customer_id=42 total_spend=10  insert customer_id=42 total_spend=10
  2  update customer_id=42 total_spend=27  update customer_id=42 total_spend=28
     ^ total_spend: expected 27, got 28

failed: 1 of 1 fixture

Exit codes: 0 when every fixture passes, 1 when any fails, 2 when the project cannot be tested at all (it does not load, check would report errors, or --model names a model that does not exist).

Determinism

The simulator is single-threaded and has no wall clock. The same fixture and the same plan produce the same changelog on every run; the repository’s own test suite runs the acceptance fixture a hundred times and asserts identical output. That is what makes a fixture a specification rather than a flaky observation, and what lets the simulator serve as the oracle the Flink backend is compared against.

Practical notes

• A project with no tests/ directory passes honestly: ok: no fixtures under tests/.
• Fixtures are checked before they run. streamform check compiles every model and validates every expectation’s shape against the model’s output columns, so a typo fails fast with the column list in the message.
• The - event: {…} form for an item in given reads well beside - watermark: … items, which advance an event-time source’s watermark explicitly. See below.

Late events and time

On a source that declares event_time, a fixture also controls the clock. Timestamps are RFC 3339 instants (2026-08-30T10:00:00Z); - watermark: <timestamp> items advance the watermark between events; and an event that arrives behind the watermark is late — dropped and counted under the default late_events: drop, and then it must be listed under late:, exactly and in order, or the fixture fails:

model: customer_metrics

given:
  - event: { customer_id: 42, amount: 10, ordered_at: "2026-08-30T10:00:00Z" }
  - event: { customer_id: 42, amount: 20, ordered_at: "2026-08-30T10:10:00Z" }
  - event: { customer_id: 42, amount: 5,  ordered_at: "2026-08-30T10:01:00Z" }   # behind the 10:05 watermark

expect:
  - { op: insert, customer_id: 42, total_spend: 10 }
  - { op: update, customer_id: 42, total_spend: 30 }

late:
  - { customer_id: 42, amount: 5, ordered_at: "2026-08-30T10:01:00Z" }

2 events processed
2 changelog records emitted
1 late event dropped

Leave the late: section out and the failure names the event and the two ways to resolve it — acknowledge it, or set late_events: keep on the source. A model with retention expires a key’s state once the watermark passes its last event plus the retention, silently; a fixture shows it by the key’s next event arriving as an insert rather than an update. The three canonical fixtures — a late event, the watermark advancing, state expiring — are walked through in Event time.

Applications of many models

A model reads another model with ref('<model>'), and Streamform plans the whole graph as one application. order-pipeline in the examples archive is the reference: three models in a chain, ending in a sink.

The example

-- models/clean_orders.sql
SELECT customer_id, amount
FROM source('orders')
WHERE customer_id IS NOT NULL

-- models/customer_metrics.sql
SELECT customer_id, SUM(amount) AS total_spend
FROM ref('clean_orders')
GROUP BY customer_id

-- models/big_customers.sql
SELECT customer_id, total_spend
FROM ref('customer_metrics')
WHERE total_spend > 100

# streamform.yml (models section)
models:
  customer_metrics:
    materialized: upsert
    key: [customer_id]
  big_customers:
    materialized: upsert
    key: [customer_id]
    sink:
      connector: kafka
      topic: big-customers
      format: json

streamform graph prints the application:

order-pipeline

orders                   source                kafka topic=orders
└─ clean_orders          append
   └─ customer_metrics   upsert(customer_id)
      └─ big_customers   upsert(customer_id)   → kafka topic=big-customers

--format dot emits Graphviz for dot -Tsvg.

What crosses a ref()

ref('m') binds to m’s Sink: its output columns and its changelog mode. That mode is what makes the last model interesting. customer_metrics emits updates, so big_customers is a filter over an updating input, and its changelog has inserts when a customer crosses 100, updates while they stay above it, and a delete when a refund takes them under. Its fixture asserts exactly that; see Testing with fixtures.

streamform explain big_customers shows the upstream model on the first operator:

big_customers

Filter
  from=customer_metrics
  predicate=(total_spend > 100)
  boundedness=Unbounded
  changelog=Upsert(customer_id)
  state=Stateless

explain with no model prints every model in dependency order.

What check catches

The dependency graph is validated before anything is planned, and each problem names the models involved:

• a ref() to a model that does not exist (with a suggestion when a close name exists);
• a model that references itself, or a cycle, listed in full;
• a name used by both a source and a model, since the two share one namespace;
• a plain table name in FROM, which would bypass the graph.

Fixtures across models

A fixture for big_customers gives events to orders, the source at the top of its chain, and asserts big_customers’ changelog after clean_orders and customer_metrics have run. You test the model you care about; Streamform runs its upstream closure.

What a chain becomes on Flink

One plan for the graph means one job for the graph. Each model is a CREATE TEMPORARY VIEW built on the view of the model it reads, so the chain runs inside a single Flink job with no topic between models; only a declared sink: writes out. A model with neither a sink nor a reader is planned but nothing on Flink runs it, and build warns.

Limits today

A model reads exactly one relation: no joins or unions yet. GROUP BY over an updating model is refused, because folding an update into an accumulator needs retractions the runtime does not produce yet. Both are on the roadmap.

Running on Apache Flink

The Flink backend runs the same plan the simulator runs. It generates Flink SQL, which you can read and diff, and submits it through Flink’s SQL Gateway. Streamform never links a Flink client and generates no Java or Python.

Declare a target

A target is a place the application is built for and applied to: a SQL Gateway and the Kafka it reads and writes.

# streamform.yml
targets:
  local:
    backend: flink
    flink:
      gateway: http://localhost:8083
    kafka:
      bootstrap_servers: kafka:9092
      startup: earliest

bootstrap_servers is as Flink sees it (inside the bundled Compose network that is kafka:9092); gateway is as your machine sees it. build and apply take --target NAME, defaulting when the project declares exactly one.

Build

streamform build --backend flink            # writes build/flink/application.sql and build/plan.json
streamform build --backend flink --stdout   # prints the SQL, writes nothing

For examples/customer-metrics:

CREATE TABLE `orders` (
  `amount` DECIMAL(38, 9),
  `customer_id` BIGINT
) WITH (
  'connector' = 'kafka',
  'topic' = 'orders',
  'properties.bootstrap.servers' = 'kafka:9092',
  'format' = 'json',
  'scan.startup.mode' = 'earliest-offset'
);

CREATE TEMPORARY VIEW `customer_metrics` AS
SELECT `customer_id`, `sum(amount)` AS `total_spend` FROM (SELECT `customer_id`, SUM(`amount`) AS `sum(amount)` FROM `orders` GROUP BY `customer_id`) AS t3;

CREATE TABLE `sink_customer_metrics` (
  `customer_id` BIGINT,
  `total_spend` DECIMAL(38, 9),
  PRIMARY KEY (`customer_id`) NOT ENFORCED
) WITH (
  'connector' = 'upsert-kafka',
  'topic' = 'customer-metrics',
  'properties.bootstrap.servers' = 'kafka:9092',
  'key.format' = 'json',
  'value.format' = 'json',
  'key.json.encode.decimal-as-plain-number' = 'true',
  'value.json.encode.decimal-as-plain-number' = 'true'
);

EXECUTE STATEMENT SET
BEGIN
  INSERT INTO `sink_customer_metrics` SELECT * FROM `customer_metrics`;
END;

The rules are few and each follows from the plan: every source is a CREATE TABLE; every model is a CREATE TEMPORARY VIEW built from its operator chain; a model with a sink: block also gets a sink_<model> table and an INSERT INTO in the single statement set. An upsert model uses upsert-kafka with a primary key on its declared key, so a delete in the changelog is a tombstone on the topic. A file sink cannot carry an upsert changelog and is refused with exit code 1. The Flink backend reference lists the type mapping and every rule.

Apply

streamform apply --target local

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

apply builds, opens a gateway session, runs each statement in order, prints the job id, and closes the session. If Flink rejects a statement, apply prints the statement it was on and Flink’s reason without the stack trace, stops with exit 1, and no job is started. An unreachable gateway is exit 2.

apply submits and reports; it does not remember the job. Stopping, upgrading, and reconciling a running application are not built yet; Preparing for upgrades covers what exists today toward that.

A local cluster in one command

The examples archive ships flink-local/docker-compose.yml: one Kafka broker (KRaft) and a Flink 2.0.2 cluster (JobManager, TaskManager, SQL Gateway) with the Kafka SQL connector on its classpath. From the streamform-examples directory:

docker compose -f flink-local/docker-compose.yml up -d --build --wait
docker exec streamform-kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic orders
streamform apply --project customer-metrics
printf '%s\n' '{"customer_id": 42, "amount": 10}' '{"customer_id": 42, "amount": 18}' \
  | docker exec -i streamform-kafka /opt/kafka/bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic orders
docker exec streamform-kafka /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic customer-metrics --from-beginning --property print.key=true
docker compose -f flink-local/docker-compose.yml down -v

The consumer prints two records for key {"customer_id":42} with totals 10 and 28, written at the column’s scale as 10.000000000 and 28.000000000: the simulator’s changelog, on Kafka. JSON sinks always write a decimal as a plain number ('json.encode.decimal-as-plain-number'); left to its default, Flink would write 10 as 1E+1. The Flink UI is at http://localhost:8081.

Two things the real cluster teaches

• Create source topics before apply. Flink’s Kafka source lists a topic’s partitions at startup and fails if there are none; do not rely on the first produced event to create the topic.
• A filter over an aggregate runs in Flink’s retract mode. An update to a row that stays inside the filter reaches an upsert-kafka topic as a tombstone followed by the new value for the same key, where the simulator’s changelog has one update. The keyed state a consumer holds is identical once both have arrived; a consumer that reacts to every record will see the key vanish and reappear.

The differential test

In the source repository, mise run integration brings the environment up and runs the differential test: for each example fixture whose model declares a sink, it applies the project, produces the fixture’s given events to the source topic, reads the sink topic, and compares the (key, value | tombstone) sequence with what the simulator’s changelog implies. Both backends run the same plan, so they must agree, and big_customers proves the delete arrives as a tombstone. The tests are #[ignore] so the ordinary check never needs Docker; CI runs them on demand and weekly.

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.

Commands

Every command takes --project DIR (default: the current directory) and compiles the whole project to one streaming plan before doing its job. Exit codes follow one convention: 0 success, 1 the command ran but the project or the fixtures are not valid, 2 the command itself could not run (the project does not load, a model is unknown, a service is unreachable).

streamform check          Load and validate a Streamform project
streamform explain        Print the streaming plan for a model
streamform test           Run every fixture in tests/ through the simulator and assert its changelog
streamform graph          Print the model dependency graph: sources, models, and sinks
streamform build          Generate the application for a backend, without submitting it
streamform apply          Build the application and submit it to a target's Flink SQL Gateway
streamform inspect-plan   Print the stable identity of the plan: node ids, digests, schemas, and contracts
streamform plan           Compare the running plan with the desired plan and classify every change

streamform check

streamform check [--project DIR] [--strict] [--json]

Loads the project, validates every file, builds the dependency graph, compiles every model, validates targets, and checks every fixture’s expectations against its model’s output columns. Prints a summary of models (with their materialization and file), sources, and fixtures, then each diagnostic with its file and a help: line, then a count.

--json prints the check report instead — check_version: 1, loaded, passed, strict, the project, the three tables as records, and diagnostics with severity, path, message, help — and nothing else on stdout. When the project cannot be loaded the report is still printed, with loaded: false and the reason as its one diagnostic, before the exit 2. streamform schema check-report prints its schema.

order-pipeline (schema version 1)

models
  big_customers            upsert(customer_id)      models/big_customers.sql
  clean_orders             append                   models/clean_orders.sql
  customer_metrics         upsert(customer_id)      models/customer_metrics.sql

sources
  orders                   kafka/json   topic=orders             2 columns

fixtures
  big_customers            tests/big_customers.yml  4 given, 3 expect
  clean_orders             tests/clean_orders.yml   3 given, 2 expect
  customer_metrics         tests/customer_metrics.yml 3 given, 2 expect

ok: 3 models, 1 source, 3 fixtures, 0 errors, 0 warnings

A diagnostic:

error: model `downstream` reads ref('other_model'), but there is no models/other_model.sql
  --> models/downstream.sql
  help: ref() names another model in this project

--strict treats warnings as errors. Exit 0 with no errors, 1 with errors, 2 if the project cannot be loaded at all.

streamform explain

streamform explain [MODEL] [--project DIR] [--json]

Prints one model’s plan as its operator chain, or every model in dependency order when no model is given. Each operator shows its definition, then boundedness, changelog, and state; a keyed operator adds its fingerprint and the first twelve characters of its state digest. A model reading another model shows from=<model> on its first operator.

--json prints the model’s slice of the plan manifest (manifest_version, model, nodes), the same document inspect-plan --json prints for the whole application. Exit 2 for an unknown model or a model that does not compile, with the compile error.

streamform test

streamform test [--project DIR] [--model MODEL] [--json]

Runs every fixture, or only those for --model, through the simulator and asserts the exact changelog. Prints per fixture: the model, PASS or FAIL, events processed, records emitted; a failure adds the fixture path and a side-by-side listing of expected and actual with the first difference named. A project with no fixtures prints ok: no fixtures under tests/ and exits 0.

Exit 0 when every fixture passes, 1 when any fails, 2 when the project cannot be tested (it does not load, check would report errors, or --model names a model that does not exist). See Testing with fixtures.

--json prints the test report instead: report_version: 1, passed, and one entry per fixture with model, path, passed, events_processed, changes_emitted, and rows — one per position, each with expected, actual, and problem (null when they match; the authority on whether the row passed). Values on both sides are rendered after typing against the model’s output schema, so a matching row looks matching: decimals are strings ("28", exact, trailing zeros trimmed as in fixtures), timestamps RFC 3339, integers and booleans themselves. An update’s actual carries before, the row as it was. A fixture the simulator refuses has error and no rows. With errors check would report, no document is printed and the exit is 2. streamform schema test-report prints its schema.

streamform graph

streamform graph [--project DIR] [--format text|dot] [--json]

Prints the application as a tree: each source, the models that read it, each model’s materialization, and declared sinks. --format dot emits Graphviz. --json prints the graph document: graph_version: 1, sources, models (each with materialized, key, sink, and what it reads), edges from source:<name> or model:<name> to model:<name>, the topological order, and cycles. A reference cycle is an error for the text and DOT forms (exit 2); the JSON form prints the document with the cycle in cycles and exits 1. streamform schema graph prints its schema.

order-pipeline

orders                   source                kafka topic=orders
└─ clean_orders          append
   └─ customer_metrics   upsert(customer_id)
      └─ big_customers   upsert(customer_id)   → kafka topic=big-customers

streamform build

streamform build --backend flink [--target NAME] [--project DIR] [--stdout]

Compiles the project (refusing anything check would fail), picks the target (required when the project declares more than one), validates the application for the backend, and writes build/flink/application.sql and build/plan.json. --stdout prints the SQL and writes nothing. A model with neither a sink nor a reader produces a warning. Exit 1 when the backend cannot represent the application (the capability error is printed), 2 when the project does not pass check or the target is unknown. See Running on Apache Flink.

streamform apply

streamform apply [--target NAME] [--project DIR]

Builds the application and submits each statement in order to the target’s Flink SQL Gateway: opens a session, runs each CREATE TABLE and CREATE TEMPORARY VIEW, runs the EXECUTE STATEMENT SET, prints the job id, closes the session.

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

Before applying, apply compares the desired plan with the target’s applied record and prints the verdict (plan: SAFE against target local, applied 2026-08-27T10:12:03Z); when the verdict is above COMPATIBLE it prints the whole diff and applies anyway, saying so, because it cannot stop or restore a job yet. After the last statement is accepted it writes the applied record, .streamform/applied/<target>.json, atomically; a rejected apply leaves the previous record.

If no model declares a sink there is nothing to run: nothing to apply: no model declares a sink. If Flink rejects a statement, apply prints the statement and Flink’s reason, stops, and exits 1; no job is started and the record is unchanged. An unreachable gateway is exit 2; so is a record that could not be written after a successful apply, and the message says the apply succeeded. The gateway is plain HTTP.

streamform inspect-plan

streamform inspect-plan [MODEL] [--project DIR] [--json]

Prints the stable identity of one model, or of every model in dependency order followed by the application digest. Per model: materialization, declared key, output node. Per node: id, operator, canonical definition, output schema with its digest, boundedness and changelog contract, and state (stateless, or keyed with key, accumulators, retention, and state digest). --json prints the plan manifest: the whole application, or one model’s slice when a model is given. Exit 2 for an unknown model or a model that does not compile. See Upgrading a running application and the plan identity reference.

streamform plan

streamform plan [--target NAME] [--against FILE] [--project DIR] [--json]

Compares the running plan with the desired plan and classifies every change. The desired plan is compiled from the project, through the same checks as build. The running plan is the target’s applied record (.streamform/applied/<target>.json, written by apply) or, with --against, any file holding a plan manifest or an applied record, such as a build/plan.json from an earlier build. --target defaults when the project declares exactly one.

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 a5771fc8a5b2

customer_metrics  changed
  customer_metrics::aggregate::0  Aggregate  changed
    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
  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

Per model, each changed node with the fields that differ (old → new), its classification (SAFE, COMPATIBLE, STATE MIGRATION REQUIRED, BACKFILL REQUIRED, STATE INCOMPATIBLE) and the reason; then the model’s verdict, the application’s verdict with counts, and a note when the target’s backend is Flink and the verdict is above SAFE. With nothing recorded and no --against, every model is new and the verdict is SAFE. --json prints the diff document (diff_version: 1). A hint in streamform.yml that names no running node is printed as a warning.

Exit 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 (a record or manifest from another format version, unknown fields, a file that is neither). See Upgrading a running application and the plan diff reference.

streamform schema

streamform schema --list
streamform schema <document> [--version N]

Prints the JSON Schema (draft 2020-12) of a document this build reads or writes. The schema is generated from the same types the binary uses, so it cannot disagree with what check accepts or what plan --json prints. --list names every document with the version this build speaks and where the document appears:

project         1  streamform.yml
sources         1  sources.yml
fixture         1  tests/*.yml
plan-manifest   1  inspect-plan --json, explain --json, build/plan.json
applied-record  1  .streamform/applied/<target>.json
diff            1  plan --json
check-report    1  check --json
test-report     1  test --json
graph           1  graph --json

--version N is refused, with exit 2, when this build speaks another version of the document; use it in scripts to assert the version you were written against. Every schema carries an $id of the form urn:streamform:spec:<document>:v<N>.

Editors that use the YAML language server (VS Code with the YAML extension, Zed, Neovim) complete and validate a project file from the schema when the file names it:

# yaml-language-server: $schema=.streamform/schema/project.json
version: 1
name: customer-metrics

Write the file once with streamform schema project > .streamform/schema/project.json.

Streamform — Project format

An Streamform project is a directory. Every file in it is a user-facing contract: this document is the reference for what each file may contain, and any change to these formats is a documented change.

streamform.yml          project name, schema version, per-model materialization
sources.yml        external inputs and their schemas (optional)
models/*.sql       one SQL model per file; the file stem is the model name
tests/*.yml        fixtures, one per file

streamform.yml

name: customer-metrics
version: 1

models:
  customer_metrics:
    materialized: upsert
    key:
      - customer_id

| targets.<name>.backend | for build/apply | The execution backend: flink. | | targets.<name>.flink.gateway | for flink | The SQL Gateway REST endpoint streamform apply submits to, e.g. http://localhost:8083. | | targets.<name>.kafka.bootstrap_servers | when Kafka is used | The bootstrap.servers list the generated source and sink tables connect to, as seen from Flink (inside the Compose network that is kafka:9092). | | targets.<name>.kafka.startup | no | earliest (default) or latest: where a Kafka source starts reading when the application first runs. |

A model exists because a models/<name>.sql file exists. An entry under models: only configures it; a model without an entry is append with no key.

A target is a place the application is built for and applied to: a Flink SQL Gateway and the Kafka it reads and writes. streamform build --backend flink and streamform apply take --target NAME, defaulting when the project declares exactly one. See flink-backend.md.

targets:
  local:
    backend: flink
    flink:
      gateway: http://localhost:8083
    kafka:
      bootstrap_servers: kafka:9092

sources.yml

sources:
  orders:
    connector: kafka
    topic: orders
    format: json
    schema:
      customer_id: int64
      amount: decimal

A bare decimal is decimal(38,9). decimal(precision,scale) declares the total number of digits (1 to 38) and the number of fractional digits (0 to the precision); the declared scale is what a model’s state keeps (a SUM over decimal(10,2) accumulates decimal(38,2)), so choose it deliberately.

models/*.sql

One SELECT statement per file, reading a source with source('<name>') or another model of the same project with ref('<model>'). The supported surface is: column references, literals, comparison and arithmetic operators, AND/OR/NOT, IS [NOT] NULL, CAST, WHERE, GROUP BY over columns, HAVING, and the aggregates sum, count, min, max. Anything else fails at streamform check naming the construct.

ref('m') reads model m’s output: its columns, and its changelog mode. A model reading an upsert model receives updates, and a WHERE over them emits an insert when a row crosses into the condition and a delete when it leaves; streamform explain shows the upstream model as from=m on the first operator. A model may read one relation only (no joins or unions yet), and GROUP BY over an updating model is refused, because folding an update into an aggregate needs retractions that nothing supports yet.

Models form a dependency graph. streamform check reports a ref() to a model that does not exist, a reference cycle, and a name used by both a source and a model (the two share one namespace). Sources are read with source(), models with ref(); a plain table name is not valid.

tests/*.yml

A fixture delivers events to one model and asserts the exact changelog the model emits.

model: customer_metrics

given:
  - customer_id: 42
    amount: 10
  - customer_id: 42
    amount: 18

expect:
  - op: insert
    customer_id: 42
    total_spend: 10
  - op: update
    customer_id: 42
    total_spend: 28

model

The model under test. It must exist.

given

The input events, delivered in order to the source at the top of the model’s chain: for a model that reads ref('clean_orders'), which reads source('orders'), events are orders rows and the changelog asserted is the model’s, after every model in between. Each event is a map of column name to value. Columns the source declares but the event omits are null; a column the source does not declare is an error.

Values are typed against the source schema: an integer or a decimal number fits a decimal column (0.1234 against decimal(38,2) is an error, not a rounding), a whole number fits an integer column, a quoted string fits a string column, true/false fits a boolean column, and a timestamp column takes an RFC 3339 instant in UTC (2026-08-30T10:00:00Z, optionally with a fraction) or an integer of milliseconds since the epoch. A bare time of day is refused.

An item may also be written as - event: {…}, which reads well beside the other kind of item: - watermark: <timestamp> advances the source’s watermark to that instant explicitly (never backwards). Watermark items mean something only on an event-time source; see Event time.

expect

The changelog the model must emit, record by record, in order. The comparison is exact: the sequences must have the same length, and record i must have the op and every column value of expect[i].

Each record has:

• op — insert, update, or delete. An update is asserted by its after-image: the values after the change.
• one entry per output column of the model, no more and no fewer. streamform check reports an expectation that misspells or omits a column before anything runs.

Values compare by type: an integer 28 matches a decimal 28.000000000; null matches only null.

late

On an event-time source with late_events: drop, the events that must be dropped as late, in delivery order, exactly — a dropped event that is not listed fails the fixture, and a listed event that was delivered fails it too. Each entry is the event as written in given. Absent when nothing is dropped. streamform test prints N late events dropped whenever N is not zero.

Running

streamform graph prints the application as a tree (sources, the models that read them, and declared sinks); streamform graph --format dot emits Graphviz for dot -Tsvg. streamform test runs every fixture; streamform test --model NAME runs the fixtures for one model. Each fixture prints its model name, PASS or FAIL, the number of events processed, and the number of changelog records emitted. A failure adds the fixture path and a side-by-side listing of expected and actual records, with the first difference named beneath each mismatching row:

customer_metrics

FAIL
  --> tests/customer_metrics.yml

2 events processed
2 changelog records emitted

  #  expected                              actual
  1  insert customer_id=42 total_spend=10  insert customer_id=42 total_spend=10
  2  update customer_id=42 total_spend=27  update customer_id=42 total_spend=28
     ^ total_spend: expected 27, got 28

failed: 1 of 1 fixture

Exit codes: 0 when every fixture passes, 1 when any fails, 2 when the project cannot be tested at all — it does not load, streamform check would report errors, or --model names a model that does not exist.

The simulator is deterministic: the same fixture and the same model produce the same changelog on every run, with no external services.

Build output

streamform build --backend flink writes the generated application to build/flink/application.sql and the plan manifest to build/plan.json: a versioned JSON document recording every node’s id, definition, schema, changelog contract, and state digest (see plan identity). --stdout prints the application instead and writes nothing. build/ is output, not source; keep it out of version control.

Applied records

streamform apply writes the plan it applied to .streamform/applied/<target>.json, with the time, the Streamform version, the backend, and the job ids. streamform plan reads it back as the running plan; --against FILE compares with any manifest or record instead. The record is what apply last did on this machine, not a query of the target; whether to commit .streamform/ is the team’s choice. See the plan diff.

Schema versioning

version is required and a build of Streamform accepts exactly one value; this build accepts 1. The version is bumped only when a project that was valid before becomes invalid or means something different afterwards. Adding an optional field does not bump it.

Unknown fields are rejected everywhere. An older build reading a project written for a newer version fails on the first field it does not know, rather than silently ignoring a materialization or a key. There is no migration tooling until a second version exists.

Schemas

Every file described here has a JSON Schema this build generates from its own types: streamform schema project, streamform schema sources, streamform schema fixture (see the command reference); the documents the binary writes — check --json, test --json, graph --json, plan --json, the plan manifest, the applied record — have one too. Name it at the top of a file for editor completion and validation — # yaml-language-server: $schema=.streamform/schema/project.json — and the editor enforces the same rules check does, including unknown fields.

Streamform — Flink backend

The Flink backend runs the same Streaming IR the simulator runs, on Apache Flink. It generates Flink SQL; it does not generate Java or Python, and Streamform never links a Flink client. streamform build --backend flink writes the script and streamform apply submits it through Flink’s SQL Gateway.

What streamform build --backend flink generates

For examples/customer-metrics with the local target:

CREATE TABLE `orders` (
  `amount` DECIMAL(38, 9),
  `customer_id` BIGINT
) WITH (
  'connector' = 'kafka',
  'topic' = 'orders',
  'properties.bootstrap.servers' = 'kafka:9092',
  'format' = 'json',
  'scan.startup.mode' = 'earliest-offset'
);

CREATE TEMPORARY VIEW `customer_metrics` AS
SELECT `customer_id`, `sum(amount)` AS `total_spend` FROM (SELECT `customer_id`, SUM(`amount`) AS `sum(amount)` FROM `orders` GROUP BY `customer_id`) AS t3;

CREATE TABLE `sink_customer_metrics` (
  `customer_id` BIGINT,
  `total_spend` DECIMAL(38, 9),
  PRIMARY KEY (`customer_id`) NOT ENFORCED
) WITH (
  'connector' = 'upsert-kafka',
  'topic' = 'customer-metrics',
  'properties.bootstrap.servers' = 'kafka:9092',
  'key.format' = 'json',
  'value.format' = 'json',
  'key.json.encode.decimal-as-plain-number' = 'true',
  'value.json.encode.decimal-as-plain-number' = 'true'
);

EXECUTE STATEMENT SET
BEGIN
  INSERT INTO `sink_customer_metrics` SELECT * FROM `customer_metrics`;
END;

The rules, each derived from the plan rather than from anything Flink-specific in it:

• Every source is a CREATE TABLE. A kafka source becomes the kafka connector with the target’s bootstrap servers, the source’s topic and format, and scan.startup.mode from the target’s kafka.startup. A file source becomes the filesystem connector.

• Every model is a CREATE TEMPORARY VIEW, built from the model’s operator chain as nested subqueries. ref('m') reads the view m, so a chain of models runs inside one job without a topic in between.

• A model with a sink: block also gets a table named sink_<model> and an INSERT INTO inside the single EXECUTE STATEMENT SET. materialized: append uses the kafka connector; materialized: upsert uses upsert-kafka with PRIMARY KEY (…) NOT ENFORCED on the model’s key, so a delete in the changelog is a tombstone on the topic. A file sink cannot carry an upsert changelog and is refused.

• Types map from the source vocabulary: int32 → INT, int64 → BIGINT, float64 → DOUBLE, decimal(p,s) → DECIMAL(p, s), string → STRING, boolean → BOOLEAN, timestamp → TIMESTAMP(3).

• Identifiers are backtick-quoted, including the names Streamform gives unaliased aggregate columns (sum(amount)); the model’s own column names are what a reader sees.

• Event time (event time): a source with event_time gets WATERMARK FOR <column> AS <column> - INTERVAL '<delay>' (rendered in the largest whole unit: '5' MINUTE, '1' HOUR, '1.500' SECOND) and, for JSON, 'json.timestamp-format.standard' = 'ISO-8601'. With late_events: drop the table is created as <source>_raw and the name models read becomes a CREATE TEMPORARY VIEW keeping only rows with CURRENT_WATERMARK(<column>) IS NULL OR <column> >= CURRENT_WATERMARK(<column>). A model’s retention becomes a STATE_TTL hint on its aggregate, which Flink enforces in processing time; build and apply warn that it is approximate and wrong under replay. Flink’s ISO-8601 for TIMESTAMP(3) takes no zone designator: producers write 2026-08-30T10:00:00, not …Z.

A model with neither a sink nor a reader is planned but nothing runs it; build warns.

Output goes to build/flink/application.sql under the project, or to standard output with --stdout. Exit code 1 means the backend cannot represent the application (a capability error is printed); 2 means the project itself does not pass streamform check.

streamform apply

streamform apply [--target NAME] builds the application and submits each statement, in order, to the target’s Flink SQL Gateway over its REST API: it opens a session, runs every CREATE TABLE and CREATE TEMPORARY VIEW, runs the EXECUTE STATEMENT SET, prints the job id Flink assigns, and closes the session.

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

If Flink rejects a statement, apply prints the statement it was on and Flink’s reason (the message lines of its exception, without the stack trace), stops, and exits 1; no job is started. A gateway that cannot be reached is exit 2. The gateway address comes from the target’s flink.gateway; it is plain HTTP, and the Flink cluster must have the Kafka SQL connector jar on its classpath for the generated tables to resolve.

Kafka topics a source reads must exist before apply: Flink’s Kafka source lists a topic’s partitions when the job starts and fails if there are none. Create them first (kafka-topics.sh --create) rather than relying on the first produced event to create them.

One Flink characteristic to know when consuming an upsert topic: a filter over an aggregate (a WHERE in a model that reads an upsert model, or HAVING) runs in Flink’s retract mode, and the upsert-kafka sink writes each retraction as a tombstone. An update to a row that stays inside the filter therefore reaches the topic as a tombstone followed by the new value for the same key, where the simulator’s changelog has one update. The keyed state a consumer holds is identical once the pair has arrived; a consumer that reacts to every record will see the key vanish and reappear.

apply submits and reports; it does not remember the job. Stopping, upgrading, and reconciling a running application are not built yet.

The integration environment

integration/docker-compose.yml runs one Kafka broker (apache/kafka:3.9.2, KRaft) and a Flink 2.0.2 cluster (JobManager, TaskManager, SQL Gateway) built from integration/flink/Dockerfile, which adds flink-sql-connector-kafka 4.0.1-2.0 to Flink’s classpath. Inside that network Kafka is kafka:9092, which is what the examples’ local target names; the host reaches the SQL Gateway at http://localhost:8083, the Flink UI at http://localhost:8081, and Kafka at localhost:9094. A release publishes both files in streamform-examples.tar.gz as flink-local/, next to the example projects, so the commands below work outside this repository with flink-local/docker-compose.yml and --project customer-metrics.

docker compose -f integration/docker-compose.yml up -d --build --wait
docker exec streamform-kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic orders
streamform apply --project examples/customer-metrics
printf '%s\n' '{"customer_id": 42, "amount": 10}' '{"customer_id": 42, "amount": 18}' \
  | docker exec -i streamform-kafka /opt/kafka/bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic orders
docker exec streamform-kafka /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic customer-metrics --from-beginning --property print.key=true
docker compose -f integration/docker-compose.yml down -v

mise run integration does that around the differential test (crates/streamform-cli/tests/integration.rs): for each example fixture whose model declares a sink, it applies the project, produces the fixture’s given events to the source topic, reads the sink topic, and compares the (key, value | tombstone) sequence with what the simulator’s changelog implies, reading Flink’s tombstone-then-value pair for one key as the single update it is. Both backends running the same IR must agree; big_customers proves the delete arrives as a tombstone. The tests are #[ignore], so mise run check never needs Docker; .github/workflows/integration.yml runs them on demand and weekly.

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.

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’.

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:

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:

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.

Streamform — Event time

Every model runs in arrival order by default: an event is processed when it shows up, and “when it happened” is a column like any other. Declaring event time on a source gives the plan a clock — a column that says when each event happened, a watermark that says how far out of order events may arrive, a rule for events that arrive later than that, and a way for keyed state to expire in event time. Fixtures can say all of it deterministically, and the same declarations run on Flink.

This page walks through the declarations, then through three fixtures that show exactly what happens — a late event, the watermark advancing, and state expiring — and ends with what Flink does with each and where it can only approximate.

Declaring it

Event time is a property of the data, so it is declared once, on the source, in sources.yml:

sources:
  orders:
    connector: kafka
    topic: orders
    format: json
    schema:
      customer_id: int64
      amount: decimal(12,2)
      ordered_at: timestamp
    event_time: ordered_at        # a timestamp column of the schema
    watermark: 5m                 # bounded out-of-orderness; default 0s
    late_events: drop             # drop (default) or keep

A source without event_time runs on processing time — exactly the behaviour before this page existed — and declaring watermark or late_events on it is an error. streamform check also rejects an event_time that is not a column or not a timestamp.

A model that keeps state can declare how long to keep it, in streamform.yml:

models:
  customer_metrics:
    materialized: upsert
    key: [customer_id]
    retention: 30d

A key’s state is dropped once the watermark reaches that key’s latest event time plus the retention. Latest means the largest event time among the events that touched the key, not the time of whichever event arrived last: an event that arrives out of order, inside the watermark delay or kept by late_events: keep, can extend a key’s life and never shortens it. Dropping is silent: nothing is retracted, and the next event for the key starts the aggregate over from its initial value — so in an upsert sink the key’s row is replaced by the fresh total, not deleted, and until then the sink keeps the stale row. retention needs an event-time source somewhere upstream, because there is no clock otherwise; check says so.

streamform explain shows the clock on the Source and the time attribute on every node that carries one; inspect-plan shows the same in the Source’s definition and a time field, and the retention in the state fingerprint:

Source
  name=orders
  connector=kafka
  time=event(ordered_at) watermark=5m late=drop
     ↓
Filter
  predicate=(customer_id IS NOT NULL)
  time attribute=ordered_at
     ↓
Aggregate
  key=customer_id
  aggregates=sum(amount)
  fingerprint=aggregate key=[customer_id] accumulators=[sum(amount): decimal(38,2)] retention=30d

Timestamps and time in fixtures

A timestamp value in a fixture is an RFC 3339 instant in UTC — 2026-08-30T10:00:00Z, with an optional fraction — or an integer of milliseconds since the epoch. A bare time of day ("10:00:00") is refused: it has no date and would make the fixture depend on the day it runs.

Two things join given and the fixture file:

• - watermark: <timestamp> advances the source’s watermark to that instant explicitly. It exists so a fixture can say “nothing more is coming before 10:10” without inventing an event. It never moves the watermark backwards.
• late: lists the events that must be dropped as late, in delivery order, exactly — the way expect: lists the changelog. A dropped event that is not listed fails the fixture, and a listed event that was delivered fails it too. A fixture never loses an event silently.

streamform test prints N late events dropped after the record count whenever N is not zero. Events dropped as late are not counted in events processed.

Three fixtures that show what happens

Each of these is a project under the repository’s crates/streamform-cli/tests/fixtures/time/ and runs in CI, so what this page says is what the simulator does. They share the source above and one model, SELECT customer_id, SUM(amount) AS total_spend FROM source('orders') GROUP BY customer_id.

A late event is dropped, counted, and acknowledged

model: customer_metrics

given:
  - event: { customer_id: 42, amount: 10, ordered_at: "2026-08-30T10:00:00Z" }
  - event: { customer_id: 42, amount: 20, ordered_at: "2026-08-30T10:10:00Z" }
  - event: { customer_id: 42, amount: 5,  ordered_at: "2026-08-30T10:01:00Z" }
  - event: { customer_id: 42, amount: 1,  ordered_at: "2026-08-30T10:06:00Z" }

expect:
  - { op: insert, customer_id: 42, total_spend: 10 }
  - { op: update, customer_id: 42, total_spend: 30 }
  - { op: update, customer_id: 42, total_spend: 31 }

late:
  - { customer_id: 42, amount: 5, ordered_at: "2026-08-30T10:01:00Z" }

Read it with the watermark in mind (watermark: 5m):

1. 10:00 arrives. Nothing has been seen, so nothing is late. The watermark becomes 09:55.
2. 10:10 arrives; the watermark becomes 10:05.
3. 10:01 arrives. It is behind 10:05, so it is late: dropped, counted, not delivered. The total stays 30.
4. 10:06 arrives. It is not behind 10:05, so it is delivered: 31. The watermark becomes 10:05 still (max seen is 10:10).

customer_metrics

PASS

3 events processed
3 changelog records emitted
1 late event dropped

Delete the late: section and the fixture fails, naming the event and saying how to acknowledge it — or how to keep it instead:

  #  late: listed  dropped
  1  (not listed)  amount=5 customer_id=42 ordered_at="2026-08-30T10:01:00Z"
     ^ dropped as late but not acknowledged; list it under `late:` or set `late_events: keep` on the source

With late_events: keep on the source, the same events produce a third update to 35 and nothing to acknowledge.

The watermark advances by events and by items, and never backwards

given:
  - event: { customer_id: 1, amount: 1, ordered_at: "2026-08-30T10:00:00Z" }   # watermark → 09:55
  - event: { customer_id: 1, amount: 1, ordered_at: "2026-08-30T09:57:00Z" }   # not late: 09:57 ≥ 09:55
  - watermark: "2026-08-30T10:20:00Z"                                          # explicit → 10:20
  - event: { customer_id: 1, amount: 1, ordered_at: "2026-08-30T10:15:00Z" }   # late
  - watermark: "2026-08-30T10:00:00Z"                                          # behind: no effect
  - event: { customer_id: 1, amount: 1, ordered_at: "2026-08-30T10:19:00Z" }   # still late
  - event: { customer_id: 1, amount: 1, ordered_at: "2026-08-30T10:30:00Z" }   # watermark → 10:25
  - event: { customer_id: 1, amount: 1, ordered_at: "2026-08-30T10:24:00Z" }   # late
  - event: { customer_id: 1, amount: 1, ordered_at: "2026-08-30T10:26:00Z" }   # delivered

Four events are delivered and three are dropped; the fixture lists the three under late:. The rule in one line: the watermark is the later of max event time seen minus the delay and the last explicit item, and it only ever increases.

State expires after its retention, and the key starts over

With retention: 10m on the model:

given:
  - event: { customer_id: 42, amount: 10, ordered_at: "2026-08-30T10:00:00Z" }
  - event: { customer_id: 7,  amount: 3,  ordered_at: "2026-08-30T10:08:00Z" }
  - watermark: "2026-08-30T10:10:00Z"
  - event: { customer_id: 42, amount: 5,  ordered_at: "2026-08-30T10:11:00Z" }
  - event: { customer_id: 7,  amount: 1,  ordered_at: "2026-08-30T10:12:00Z" }

expect:
  - { op: insert, customer_id: 42, total_spend: 10 }
  - { op: insert, customer_id: 7,  total_spend: 3 }
  - { op: insert, customer_id: 42, total_spend: 5 }     # not update 15: the key started over
  - { op: update, customer_id: 7,  total_spend: 4 }

Key 42’s last event is at 10:00; when the watermark reaches 10:10 — exactly last event plus retention, the edge is inclusive — its state is dropped. Nothing is emitted for that. The next event for 42 is therefore an insert of 5, not an update to 15. Key 7’s last event is at 10:08, so it survives the same watermark and gets an ordinary update.

What this means for a consumer of an upsert sink: between the expiry and the next event, the topic still carries 42 → 10; then it carries 42 → 5. The stale row is replaced, never deleted. If a delete on expiry matters to you, that is a native-runtime capability the roadmap lists, not something Flink SQL can express, and Streamform does not pretend otherwise.

What Flink does with each

streamform build --backend flink turns the declarations into Flink SQL, and the integration test proves Flink and the simulator agree on in-order events:

Four honest limits, stated here so nobody discovers them in production:

• Retention on Flink is approximate. The simulator expires state in event time; Flink’s hint expires it on the wall clock. Day to day they agree. Under replay they do not: thirty days of events replayed in an hour expire nothing on Flink, so a backfill can hold state a fresh run would have dropped. build and apply print a warning on every model that declares retention.
• Lateness is deterministic in the simulator and only eventually consistent on Flink, which generates watermarks periodically and per partition. The differential test covers event time and watermarks with in-order events and does not assert which events Flink drops. - watermark: items are simulator-only.
• Flink’s ISO-8601 for TIMESTAMP(3) takes no zone designator. A producer writing JSON to a topic an event-time source reads must write 2026-08-30T10:00:00 (optionally .123), not …Z. Fixtures still take the Z form, because a fixture is an instant; the integration test drops the Z when it produces to Kafka.
• An event without its event time is refused by the simulator, and Flink is not asserted. streamform test fails the fixture and names the event (event 2 has no `ordered_at` ); it is not treated as late, and not delivered at the watermark. What Flink does with a row whose time column is null is not covered by the integration run, so keep such rows off an event-time topic.

What changes in identity and the diff

Declaring retention fills the retention= of a model’s state fingerprint, which has said unbounded since Phase 1, and so moves its state digest — and only that digest. streamform plan against a running plan then reports the aggregate as STATE MIGRATION REQUIRED: timers change, stored values do not, the classification that was unreachable until now. Declaring event_time moves no state or schema digest: the source’s manifest definition gains a suffix and every node that carries the attribute gains a time field, and a project without event time produces a manifest byte-identical to one written before event time existed.

That does not make an edit to a clock free. The source’s definition is part of what the state downstream was accumulated under, so streamform plan never reports one of these as SAFE:

The Source node itself is stateless and reports COMPATIBLE; a chain with no keyed state below the source stays COMPATIBLE. The rule is conservative on purpose: a longer delay on a topic that never saw a late event changes nothing in practice, and the diff cannot know that. On Flink the STATE_TTL hint is part of the query, so a retention edit is a changed query there like any other (plan diff).

Not here yet

Windows (which consume the watermark this page produces), joins and watermarks across several sources, a processing-time column or watermark, retraction on expiry, an idle-source timeout, time zones beyond UTC instants. The roadmap says where each stands; tumbling windows are next.

The protocol: what every --json document promises

Streamform exchanges documents with everything outside it: the project files it reads, the plan documents it writes, and the reports its commands print with --json. Each is a published specification — a JSON Schema generated from the Rust type that owns it, with prose and real examples, under spec/ in the source tree and mirrored to streamform-spec on every release. These are the rules every document and every command follow; they are the contract a tool, a CI job, or a library can build on without Streamform’s source.

The documents

streamform schema --list prints the same table for the build you have, with the version each document is at.

The rules

1. Every document has a version field, first. It is named <document>_version (the three project files share version in streamform.yml). Every document started at 1.
2. A version is bumped only for a change an older Streamform must refuse. A field an older build may ignore is added with a default and no bump. A change in meaning, a removed field, a renamed one — bumps. When in doubt, it bumps.
3. A newer version is refused, never guessed. The error names the version found and the version this build speaks. A document with fields this build does not know is refused too (additionalProperties: false in every schema), so a document from a newer build that only added a field fails loudly instead of being half-read.
4. Exit codes are 0, 1, 2 for every command. 0: the command ran and found nothing wrong. 1: the command ran and found something — problems in check, a failed fixture in test, a verdict above COMPATIBLE in plan, a cycle in graph --json. 2: the command could not do its job — the project does not load, a file cannot be read, a target is missing. Gate CI on 1; 2 is a setup problem.
5. With --json, stdout holds the document and only the document. Anything written for a person goes to stderr. One exception: check --json prints a document even when the project fails to load — loaded: false, the reason as its one diagnostic, then exit 2 — because saying what is wrong is check’s job.
6. Digests are lowercase-hex SHA-256 wherever a document carries one: the application digest, schema digests, state digests.
7. Values in reports are rendered after typing. In the test report, both the expected and the emitted value are shown as the model’s column type renders them — decimals as strings (exact; trailing zeros trimmed as in a fixture), timestamps as RFC 3339 in UTC, integers and booleans as themselves — so a matching row looks matching. problem is the authority on whether it matched.

Using the schemas

• Editors. Put # yaml-language-server: $schema=.streamform/schema/project.json at the top of streamform.yml (and likewise for sources.yml and fixtures) after streamform schema project > .streamform/schema/project.json; any editor with the YAML language server then completes field names and flags unknown ones the way check would.
• CI. Parse test --json and gate on .passed; parse plan --json and gate on .verdict. Assert the version you were written against with streamform schema <document> --version N, which exits 2 if the build speaks another.
• Programs. Validate any document against its schema with a JSON Schema 2020-12 validator; every schema carries an $id of the form urn:streamform:spec:<document>:v<N>.

The published copy: glyf-data/streamform-spec, one directory per document, v<N>/schema.json beside v<N>/examples/ and a README.md of prose. Every file in it is generated from the source tree; requests go to the issue tracker, not as pull requests to the mirror.

Changelog

All notable changes to Streamform are recorded here. The format follows Keep a Changelog and the project uses Semantic Versioning; until 1.0, a minor version may contain breaking changes, and each is marked Breaking in its entry.

Each release section is copied verbatim into the GitHub release notes by the release workflow, so write entries for the people who will run the release, not for the people who wrote it.

Unreleased

Added

• Event time on Flink (DEC-009 decision 6): a source with event_time gets WATERMARK FOR <column> AS <column> - INTERVAL '<delay>' SECOND in its CREATE TABLE and, for JSON, 'json.timestamp-format.standard' = 'ISO-8601'; with late_events: drop the table is created as <source>_raw and the name models read becomes a view that keeps only rows at or after CURRENT_WATERMARK; a model’s retention becomes a STATE_TTL hint on its aggregate, and build and apply warn that on Flink it is enforced in processing time — approximate, and wrong under replay. JSON timestamps on Kafka are read in Flink’s ISO-8601 form, yyyy-MM-ddTHH:mm:ss[.SSS] without a zone. The integration run gains a differential test over an event-time source with in-order events; lateness and expiry are not asserted there.
• Event time runs in the simulator (DEC-009 decisions 3 and 5): a source with event_time keeps a watermark — the largest event time seen minus its watermark delay, never moving backwards, or a fixture’s - watermark: <timestamp> item, whichever is later. An event behind the watermark when it is delivered is late: with late_events: drop it is dropped and counted, with keep it is delivered as if on time. A model’s retention expires a key’s state once the watermark reaches its last event time plus the retention — silently, so the next event for the key starts over as an insert and an upsert sink keeps the stale row until then. streamform test prints N late events dropped, and the fixture’s late: list must name every dropped event, exactly and in delivery order: a dropped event that is not listed fails the fixture with a message saying how to acknowledge it, as does a listed event that was delivered. events processed counts delivered events. test --json gains late_dropped and late rows per fixture (defaulted fields, report_version stays 1).
• Event time is declared and carried: a source in sources.yml may name its event_time column (a timestamp), its watermark (bounded out-of-orderness, a duration such as 5m, default 0s), and late_events (drop or keep, default drop); a model may declare retention (a duration, enforced in event time by the simulator; see the entry above). explain prints time=processing or time=event(<column>) watermark=… late=… on every Source and time attribute=<column> on every node that carries one; inspect-plan prints the same in the Source’s definition and a time field, which is omitted when there is none, so a plan without event time is byte-identical to one written by v0.4.0 and every recorded digest stands. A model’s retention appears in its state fingerprint’s retention= (which has said unbounded since Phase 1) and therefore moves its state digest and only that digest. check rejects an event_time that is not a timestamp column, a watermark or late_events without event_time, and a retention on a model with no event-time source upstream. Fixtures accept RFC 3339 timestamps (2026-08-30T10:00:00Z) for timestamp columns beside the integer form, - watermark: <timestamp> items in given, and a late: list (DEC-009).
• Every release publishes streamform-examples.tar.gz beside the binaries, covered by checksums.txt: the example projects, and the local Kafka and Flink environment as flink-local/. The source repository is private, so this is how the documented first project and the Flink guide are reproduced from public artifacts; the README, the install page, and the guides download it instead of cloning.
• The formats are published: spec/ in the repository holds, for every document the binary reads or writes, its JSON Schema (generated from the binary with mise run spec; CI fails when it drifts), real examples produced from the example projects, and a page of prose; the whole directory is mirrored to glyf-data/streamform-spec on every release. docs/protocol.md states the rules every document and every --json follow: a version field first, newer versions and unknown fields refused, exit codes 0/1/2, stdout holds only the document, values rendered after typing.
• streamform check --json, streamform test --json, and streamform graph --json print one versioned document each (check_version, report_version, graph_version, all 1) and nothing else on stdout; exit codes are unchanged. The test report carries every comparison row the text output shows, with values on both sides rendered after typing against the model’s schema — decimals as strings, so nothing is lost to a float, timestamps as RFC 3339 — and an update’s before-image. check --json prints a document even when the project fails to load (loaded: false, exit 2); graph --json prints the document with its cycles on a reference cycle and exits 1, where the text form refuses.
• streamform schema <document> prints the JSON Schema (draft 2020-12) of a document this build reads or writes — project, sources, fixture, plan-manifest, applied-record, diff, check-report, test-report, graph — generated from the same types the binary uses, so the schema cannot disagree with the binary; streamform schema --list names every document with its version and where it appears. --version N is refused when this build speaks another version. A project file can name its schema for editor completion with # yaml-language-server: $schema=<file>.

Fixed

• Retention counts from the latest event time a key has seen. The simulator counted from whichever event arrived last, so an out-of-order event (inside the watermark delay, or kept by late_events: keep) could bring a key’s expiry forward and drop state that a more recent event had just touched. A write now extends a key’s life or leaves it alone, as it does under Flink’s state TTL.
• A fixture event with no value in the event-time column fails with event N has no `<column>`; source `<source>` runs on event time, so every event needs one, where it used to say plan cannot be simulated. Messages for constructs the simulator does not run say is not supported yet, not in v0.1.
• JSON sinks on Flink write decimals as plain numbers. Flink’s JSON format strips trailing zeros and falls back to scientific notation by default, so a total of 10 reached Kafka as 1E+1, 100 as 1E+2, and 0.000000027 as 2.7E-8. Generated sink tables now set 'json.encode.decimal-as-plain-number' = 'true' (for upsert-kafka, on both the key and the value format), and the same values arrive as 10.000000000, 100.000000000, and 0.000000027, at the column’s scale. Anyone parsing a sink topic as text sees different bytes for the same number; a JSON parser sees the same value. build output changes for every project with a JSON sink.

Changed

• The documentation says what exists: Homebrew, signed artifacts, a container image, and further targets are planned and not available, as are the testing library, the Flink runner image, and the GitHub Action; the only command that opens a network connection is apply, to the SQL Gateway it is given.
• examples/order-pipeline declares event time: orders gains ordered_at: timestamp, event_time: ordered_at, and watermark: 5m; every fixture event carries a timestamp, and tests/late_order.yml shows a late order dropped and acknowledged. docs/event-time.md (new) walks through event time, watermarks, late events, and retention with three runnable fixtures, and what Flink does with each; the site gains a Time concept page and the testing guide a late-events section.
• Breaking (terms): Streamform is a closed-source binary, free to use locally and in production with no account and no limit; LICENSE is now the terms of use rather than the MIT licence (no copy of the MIT-licensed tree was ever distributed). The file formats, the testing library, the Flink runner image, and the GitHub Action are open source under Apache-2.0 in their own repositories. Releases are published to glyf-data/streamform-releases; cargo install --git is no longer an install path.

0.4.0 - 2026-08-27

Streamform now says what an upgrade costs before it is deployed: streamform plan compares the running plan with the desired one and classifies every change (SAFE, COMPATIBLE, STATE MIGRATION REQUIRED, BACKFILL REQUIRED, STATE INCOMPATIBLE), streamform apply records what it applied, and streamform.yml can say that a model was renamed so the diff compares the same node.

Added

• streamform plan [--target NAME] [--against FILE] [--json] compares the running plan with the desired plan and classifies every change as SAFE, COMPATIBLE, STATE MIGRATION REQUIRED, BACKFILL REQUIRED, or STATE INCOMPATIBLE, most severe first, with the fields that differ and a reason per node. A keyed node whose upstream definition changed, even in another model, is raised to BACKFILL REQUIRED, since its state was accumulated under the previous definition. Exit 0 when the application can continue from its existing state, 1 otherwise, 2 when the comparison could not run. Documented in docs/plan-diff.md.
• streamform apply writes the plan it applied, with the time, version, backend, and job ids, to .streamform/applied/<target>.json, and prints the verdict against that record before applying again. It does not refuse on any verdict yet: the Flink backend starts a new job with empty state on every apply, and refusing waits for job lifecycle.
• renamed_from: <model> and nodes: { <kind>.<ordinal>: { was: <id> } } on a model in streamform.yml tell plan which running node a desired node was, so a renamed model or a shifted ordinal compares as the same node. streamform check rejects malformed or contradictory hints and a hint for a node the plan does not have; plan warns about a hint that names no running node.

0.3.0 - 2026-08-26

Streamform is the project’s name from this release on, and every stateful node now has an identity that survives an upgrade: streamform inspect-plan prints stable node ids, SHA-256 state and schema digests, and changelog contracts, streamform build writes them to build/plan.json, and a documentation site lives under site/. The release artifacts are the first named streamform-<target>.tar.gz.

Added

• streamform inspect-plan [MODEL] [--json] prints the stable identity of the plan: every node’s id, definition, output schema with its digest, changelog contract, and, for a keyed operator, its key, accumulators, retention, and state digest. With --json it prints the plan manifest, a versioned document (manifest_version: 1) holding the same for the whole application plus an application digest. Documented in docs/plan-identity.md.
• streamform build writes the plan manifest to build/plan.json next to the generated application, so every build carries the identity it was built from.
• A documentation site under site/ (mise run docs, mise run docs-serve): getting started, concepts, guides, and reference, with the founding documents included from docs/. A landing page lives at site/landing/index.html. The Docs workflow builds both on every change; the site is run locally, nothing deploys it.

Changed

• Breaking: the project is renamed to Streamform. The binary is streamform, the crates are streamform-*, the project manifest file is streamform.yml, and the release artifacts are streamform-<target>.tar.gz. Releases v0.1.0 and v0.2.0 shipped under the previous name; their entries below are written with the current names.
• Breaking: an aggregate column without an alias is now named by Streamform (sum(amount), count(*)) rather than by the SQL planner (sum(orders.amount)), and an aggregate column keeps its accumulator type through a projection (decimal(38,2) for a SUM over decimal(10,2), where the planner said decimal(20,2)). Generated Flink SQL changes accordingly for unaliased aggregates; fixtures that name such a column must use the new name. Aliased columns are unaffected.
• streamform explain prints a digest= line under a keyed operator’s fingerprint: the first 12 characters of the SHA-256 of the fingerprint text.
• Breaking: streamform explain --json prints the model’s slice of the plan manifest (manifest_version, model, nodes) instead of the previous hand-written shape; every field of the plan is now present, and fingerprint_digest is state.digest.

Fixed

• State fingerprint digests were computed with the standard library’s hasher, whose output may change between Rust releases. They are now SHA-256 over the fingerprint’s canonical text, so a toolchain upgrade cannot mark existing state incompatible.

0.2.0 - 2026-08-25

Streamform understands an application, not a statement, and runs it on Flink. ref() chains models into a dependency graph; streamform graph draws it; streamform build --backend flink turns the plan into Flink SQL and streamform apply submits it through the SQL Gateway; mise run integration proves on real Kafka and Flink 2.0.2 that the simulator and Flink agree. Coverage fixtures now run every aggregate, operator, and column type through SQL.

Added

• ref('<model>') in model SQL reads another model of the same project. Models form a dependency graph; streamform check reports an unknown ref(), a reference cycle, and a name used by both a source and a model. streamform explain <model> shows the upstream model as from=<model> on the first operator, and streamform explain with no model prints every model in dependency order.
• A fixture may target a model that reads other models: given events go to the source at the top of the chain and expect is asserted on the target model’s output, after every model in between.
• A model’s Sink shows its declared materialization and, when a sink: block is configured, the connector and target.
• streamform build --backend flink [--target NAME] [--stdout] generates the application as Flink SQL: a CREATE TABLE per source, a CREATE TEMPORARY VIEW per model, a CREATE TABLE sink_<model> per declared sink (upsert-kafka with a primary key for upsert models), and one EXECUTE STATEMENT SET. No Java or Python is generated. Documented in docs/flink-backend.md.
• streamform apply [--target NAME] builds the application and submits it through the target’s Flink SQL Gateway REST API, statement by statement, printing the job id; a rejected statement stops the apply with Flink’s reason and exit code 1.
• integration/docker-compose.yml runs Kafka and Flink 2.0.2 with the Kafka connector locally; mise run integration applies the examples to it and checks that the sink topics carry exactly what the simulator’s changelog implies, tombstones included.
• targets: in streamform.yml names where an application is built for and applied to: backend, flink.gateway, kafka.bootstrap_servers, kafka.startup. streamform check reports a Flink target missing its sections.
• streamform graph prints the model dependency graph as an indented tree; streamform graph --format dot emits Graphviz.
• examples/order-pipeline: the three-model example, orders → clean_orders → customer_metrics → big_customers, whose last fixture asserts an insert, an update, and a delete from one filter over an upsert model.

Changed

• Breaking for callers of the Rust API: streamform_sql::compile and compile_all are replaced by compile_application, which returns one plan for the whole project; streamform_sim::run takes the model to run. The CLI is unchanged.
• GROUP BY over a model that emits updates is refused at planning (“aggregation over an updating stream”), since aggregating updates needs retractions that are not supported yet.
• Unsupported constructs are reported as “not supported yet” rather than “not supported in v0.1”.
• An upsert model’s declared key must match the key its plan produces: streamform check now reports a key that disagrees with the GROUP BY (previously accepted silently), and a global aggregate is declared with no key (previously impossible).

Fixed

• COUNT(*) failed to plan (“the expression *”) in every release so far; nothing had exercised it through SQL. A coverage fixture project now runs every aggregate, operator, and column type through SQL.

0.1.0 - 2026-08-24

The first release: deterministic testing and inspection for streaming SQL semantics. No Flink, Kafka, dbt, Python, windows, watermarks, joins, ref(), or deployment yet.

Added

• streamform check loads and validates a project — streamform.yml, sources.yml, models/*.sql, tests/*.yml — compiles every model, and reports every problem in one run with did you mean suggestions.
• streamform explain <model> prints the model’s streaming plan: operators, boundedness, changelog mode, state requirement, key, and a state fingerprint naming each accumulator and its type. --json for machine consumption.
• streamform test [--model NAME] runs every fixture through a deterministic in-process simulator and asserts the exact changelog, record by record. Prints PASS/FAIL with event and record counts, a side-by-side diff on failure, and exits 1 when any fixture fails.
• The Streaming IR (streamform-ir): source, filter, projection, keyed aggregation, and sink operators; append and upsert changelog modes; stable node identity <model>::<kind>::<ordinal>; state fingerprints rendered through Streamform-owned type names so a dependency upgrade cannot move them.
• The SQL frontend (streamform-sql): DataFusion plans a model’s SELECT, and lowering produces the IR. Supported: column references, literals, comparison and arithmetic operators, AND/OR/NOT, IS [NOT] NULL, CAST, WHERE, GROUP BY over columns, HAVING, and the aggregates sum, count, min, max. Anything else fails naming the construct. ORDER BY over an unbounded source is rejected as meaningless.
• The simulator (streamform-sim): single-threaded, in-memory, no wall clock; operators defined over changelogs so a filter after an aggregate emits inserts and deletes at the predicate boundary; exact fixed-point decimals; checked arithmetic that reports overflow rather than wrapping.
• Project format: source('<name>') in SQL; per-model materialized: append | upsert with key; source column types int32, int64, float64, decimal, decimal(precision,scale), string, boolean, timestamp; fixtures with given events and expect records carrying op: insert | update | delete. Documented in docs/project-format.md.
• Release builds for macOS (Intel and Apple silicon) and Linux x86_64 with checksums.txt, and cargo install --git from the tag.

Streamform — Vision

What is Streamform?

Streamform is a Rust-native framework for building, testing, planning, and deploying continuously running data applications from declarative SQL models.

The user describes:

source
+
transformation
+
materialization

Streamform turns that into:

continuous dataflow

An execution backend such as Apache Flink runs that dataflow in production.

SQL models
   +
source definitions
   +
materialization
        │
        ▼
      Streamform
        │
        ▼
Streaming IR
    /       \
Simulator   Flink

Streamform itself is not another streaming database and initially is not another distributed streaming engine.

Its value sits above execution engines:

Give data engineers and software engineers a software-engineering workflow for long-running stateful data applications.

Problem

There is a large gap between writing a transformation and operating it continuously.

A batch data engineer can comfortably write:

SELECT
    customer_id,
    SUM(amount) AS total_spend
FROM orders
GROUP BY customer_id

But turning that logic into:

Kafka
   ↓
continuous processing
   ↓
persistent state
   ↓
safe updates
   ↓
production deployment

requires understanding:

• Kafka
• Flink
• state management
• changelogs
• checkpoints
• watermarks
• event time
• serialization
• deployment
• savepoints
• upgrades
• recovery

The application also behaves differently from batch SQL.

A batch query:

starts
↓
processes data
↓
finishes

A streaming application:

starts
↓
processes events
↓
accumulates state
↓
continues running
↓
changes over time
↓
must survive deployments
↓
may run for years

That lifecycle is the real problem Streamform should solve.

User Experience

A project might eventually look like:

fraud-pipeline/
├── streamform.yml
├── sources.yml
├── models/
│   ├── clean_transactions.sql
│   └── account_activity.sql
└── tests/
    └── account_activity.yml

A source:

sources:
  commerce.transactions:
    connector: kafka
    topic: transactions
    format: json

A model:

SELECT
    account_id,
    COUNT(*) AS transaction_count,
    SUM(amount) AS total_amount
FROM source('commerce.transactions')
GROUP BY account_id

A materialization:

models:
  account_activity:
    materialized: upsert

    key:
      - account_id

    sink:
      connector: kafka
      topic: account-activity

The developer workflow:

streamform check
streamform test
streamform explain
streamform plan
streamform apply
streamform inspect

Core Insight

SQL alone is not a streaming application.

A streaming application consists of:

INPUT SEMANTICS

What is the input?
Bounded file?
Kafka stream?
CDC changelog?

        +

COMPUTATION

What transformation should happen?

        +

MATERIALIZATION SEMANTICS

How should continuously changing results be represented?
Append?
Upsert?
Retract?
Queryable state?

Streamform owns the semantics connecting these three pieces.

First Principle: Streaming Is About Change

Consider:

SELECT
    customer_id,
    SUM(amount)
FROM orders
GROUP BY customer_id

Input events might be append-only:

customer=42 amount=10

customer=42 amount=18

But the output relation changes:

INSERT
customer=42 total=10

UPDATE
customer=42 total=28

Streamform must understand this distinction.

Therefore concepts such as:

Append
Upsert
Update
Retract

are first-class concepts.

They are not implementation details of Flink.

Second Principle: State Is Part of the Program

The aggregation:

GROUP BY customer_id

requires the runtime to remember previous values.

Conceptually:

state

customer 42
    count = 18
    total = 902

customer 91
    count = 4
    total = 88

Therefore Streamform must understand:

which operators require state

how that state is keyed

what schema the state has

whether state can survive a new program version

State must be represented explicitly in Streamform’s semantic model.

Third Principle: Streaming Programs Must Be Testable

A user should not need:

Kafka
+
Flink cluster
+
Docker
+
sleep(10)
+
grep logs

to test transformation logic.

Streamform should provide deterministic streaming tests.

Example:

model: customer_metrics

given:
  - customer_id: 42
    amount: 10

  - customer_id: 42
    amount: 18

expect:
  - op: insert
    customer_id: 42
    total_spend: 10

  - op: update
    customer_id: 42
    total_spend: 28

Then:

streamform test

executes completely in-process.

fixture events
      ↓
Streaming IR
      ↓
Rust simulator
      ↓
exact changelog

No external infrastructure.

Fourth Principle: Running Applications Have a Lifecycle

Streamform’s long-term differentiation should not be merely:

compile SQL to Flink.

The difficult production question is:

What happens when the SQL changes while the application already has hundreds of gigabytes of state?

Example:

Previous:

GROUP BY customer_id

New:

GROUP BY customer_id, country

Streamform should eventually tell the developer:

STATE INCOMPATIBLE

Model:
customer_metrics

Grouping key changed:

- customer_id
+ customer_id, country

Existing keyed state cannot be safely reused.

Required action:
REBUILD

Another change could result in:

SAFE

Existing state can be reused.

Or:

BACKFILL REQUIRED

This requires Streamform to understand the lifecycle of the program, not merely its syntax.

Fifth Principle: Execution Engines Are Backends

Initial architecture:

                    Streamform Streaming IR
                         /       \
                        /         \
             Simulator             Flink

Later:

                    Streamform Streaming IR
                  /         |          \
                 /          |           \
          Simulator       Flink       Streamform Runtime

Flink provides mature production execution.

Streamform owns:

• authoring semantics
• streaming semantics
• state semantics
• changelogs
• validation
• testing
• lifecycle
• plan compatibility

This prevents Streamform from becoming coupled to Flink.

Why Rust?

The implementation should be Rust-first.

Rust is particularly appropriate because the project can eventually involve:

• SQL compilation
• Arrow-native memory
• query-plan transformations
• event processing
• state stores
• async networking
• Kafka
• storage
• concurrency
• checkpointing
• distributed execution

It also lets Streamform ship as a single native CLI without requiring a Python runtime.

Initial installation should aim toward:

brew install streamform

or:

cargo install streamform

Python bindings or:

pip install streamform

can be introduced later if there is a concrete Python integration need.

Python should not be required for the initial architecture.

What Streamform Is Not

Streamform is initially not:

• a streaming database
• a Kafka replacement
• a Flink replacement
• a warehouse
• a generic workflow orchestrator
• a dbt fork
• an agent framework
• a hosted runtime (a hosted control plane that never runs user workloads is not excluded; the founding text read “a hosted SaaS platform”)
• an API framework
• a distributed Rust streaming engine

Those boundaries are deliberate.

Initial User

The initial user is an engineer comfortable with:

SQL
data models
Kafka concepts
Git
CI

but who does not want streaming application development to require deep Flink expertise.

Later users may include:

• analytics engineers
• data platform teams
• backend engineers
• real-time product teams

Longer-Term Direction

The project can evolve naturally:

Phase 1
streaming compiler + simulator

        ↓

Phase 2
production Flink execution

        ↓

Phase 3
safe application upgrades

        ↓

Phase 4
runtime contracts

        ↓

Phase 5
native Rust runtime

        ↓

Phase 6
queryable materialized state

        ↓

Phase 7
real-time API serving

Eventually:

events
  ↓
Streamform model
  ↓
continuous state
  ↓
REST / gRPC / Arrow Flight

could allow a data engineer to turn continuously maintained data into production application state without separately building an API service.

That is a future extension, not the initial objective.

Product Thesis

Streamform’s thesis is:

Continuously running data applications should be developed with the same confidence, testability, versioning, and deployment discipline as normal software.

The product is therefore not primarily:

easier streaming SQL.

It is:

a software development system for stateful streaming applications.

Streamform — Architecture

This document describes the founding architecture, with some examples from the initial implementation. For the assessed current baseline, delivery sequence, and upcoming window/lifecycle decisions, see the roadmap; the delivery work behind it is kept in the source repository (docs/implementation-strategy.md). Future-looking examples here do not imply that a command or guarantee is implemented.

Architectural Overview

                    Streamform Project
                         │
             ┌───────────┴───────────┐
             │                       │
          SQL models             metadata
                                 sources
                                 sinks
                                 tests
             │                       │
             └───────────┬───────────┘
                         ▼
                ┌────────────────┐
                │   Frontend     │
                │                │
                │ DataFusion     │
                │ LogicalPlan    │
                └───────┬────────┘
                        ▼
                ┌────────────────┐
                │   Semantics    │
                │                │
                │ boundedness    │
                │ state          │
                │ changelog      │
                │ keys           │
                │ time           │
                └───────┬────────┘
                        ▼
                ┌────────────────┐
                │ Streaming IR   │
                └───────┬────────┘
                        │
             ┌──────────┼───────────┐
             ▼          ▼           ▼
         Simulator    Flink      Native Rust
                                    future

1. Rust Workspace

Start with a small workspace.

streamform/
├── Cargo.toml
├── README.md
├── AGENTS.md
│
├── docs/
│   ├── vision.md
│   ├── architecture.md
│   └── roadmap.md
│
├── crates/
│   ├── streamform-cli/
│   ├── streamform-project/
│   ├── streamform-sql/
│   ├── streamform-ir/
│   ├── streamform-semantics/
│   └── streamform-sim/
│
└── examples/
    └── customer-metrics/

Do not create a crate for every future concept.

Add boundaries only when real code requires them.

2. Project Frontend

Streamform V0 should use its own minimal project format.

Do not make dbt a runtime dependency.

Example:

streamform.yml
sources.yml
models/*.sql
tests/*.yml

Later:

dbt manifest
     ↓
DbtFrontend
     ↓
same internal representation

becomes another frontend.

This keeps:

Streamform semantics

independent from:

dbt semantics

3. DataFusion Boundary

Use DataFusion for generic relational machinery.

SQL
 ↓
DataFusion parser/planner
 ↓
DataFusion LogicalPlan
 ↓
Streamform lowering
 ↓
Streaming IR

DataFusion provides machinery such as:

• SQL parsing
• column resolution
• expression trees
• data types
• coercion
• aggregates
• joins
• logical plans
• Arrow schemas

Streamform should not make DataFusion’s plan canonical.

Lower relatively early.

4. Streaming IR

The Streaming IR is the central architecture.

A possible starting point:

#![allow(unused)]
fn main() {
pub struct StreamingPlan {
    pub nodes: Vec<StreamNode>,
}
}

#![allow(unused)]
fn main() {
pub struct StreamNode {
    pub id: NodeId,
    pub operator: Operator,
    pub schema: Schema,
    pub boundedness: Boundedness,
    pub changelog: ChangelogMode,
    pub state: StateRequirement,
}
}

Initial operators:

#![allow(unused)]
fn main() {
pub enum Operator {
    Source(Source),
    Filter(Filter),
    Project(Project),
    Aggregate(Aggregate),
    Sink(Sink),
}
}

Keep V0 intentionally tiny.

5. Stable Node Identity

Every stateful node should have stable identity.

For example:

#![allow(unused)]
fn main() {
pub struct NodeId(String);
}

A node identity should not depend purely on:

position in DAG

because the eventual lifecycle system must compare:

Plan V1
vs
Plan V2

Stable identity enables:

state compatibility
plan diff
migration planning
deployment reconciliation

Design this early.

6. Boundedness

Streamform should explicitly model:

#![allow(unused)]
fn main() {
enum Boundedness {
    Bounded,
    Unbounded,
}
}

Examples:

Parquet snapshot
    → Bounded

Kafka topic
    → Unbounded

This changes what SQL means.

For instance:

ORDER BY event_time

over an infinite Kafka stream cannot produce a globally final ordering.

The semantic layer should catch such cases.

7. Changelog Model

The result of a streaming operator is not always append-only.

Start with something like:

#![allow(unused)]
fn main() {
enum Change {
    Insert(Record),
    Update {
        before: Record,
        after: Record,
    },
    Delete(Record),
}
}

The logical capability could separately be:

#![allow(unused)]
fn main() {
enum ChangelogMode {
    Append,
    Upsert(Key),
    Retract,
}
}

This distinction becomes foundational for:

• testing
• sinks
• materialized state
• API serving
• subscriptions

8. State Requirements

Represent state explicitly.

#![allow(unused)]
fn main() {
enum StateRequirement {
    Stateless,

    Keyed {
        keys: Vec<Column>,
        fingerprint: StateFingerprint,
    },
}
}

For:

GROUP BY customer_id

derive:

state:
    keyed

key:
    customer_id

A first simulator implementation can use:

#![allow(unused)]
fn main() {
HashMap<Key, AggregateState>
}

There is no need for RocksDB or distributed state initially.

9. State Fingerprint

Introduce the concept early, even if compatibility logic comes later.

Conceptually:

StateFingerprint

operator:
    aggregate

key:
    customer_id

accumulators:
    count: int64
    total_spend: decimal

retention:
    unbounded

Two program versions can eventually compare these fingerprints.

This powers:

streamform plan

10. Deterministic Simulator

The simulator is the first execution backend.

Architecture:

Fixture Events
      │
      ▼
   Source
      │
      ▼
   Filter
      │
      ▼
 Aggregate
      │
      ▼
  Changelog

Properties:

• single process
• initially single-threaded
• in-memory
• deterministic
• no Kafka
• no Flink
• no Docker
• no wall-clock dependency

The simulator is simultaneously:

1. a testing engine
2. the second consumer of the IR
3. the seed of the future Rust runtime

11. Test Model

Example:

model: customer_metrics

given:
  - customer_id: 42
    amount: 10

  - customer_id: 42
    amount: 18

expect:
  - op: insert
    customer_id: 42
    total_spend: 10

  - op: update
    customer_id: 42
    total_spend: 28

Then:

streamform test

produces:

customer_metrics

PASS

2 events processed
2 changelog records emitted

The fixture format — how given values are typed against the source schema, what an expect record must contain, and how a failure is rendered — is specified in project-format.md. The comparison is exact: same number of records, same op, same column values, in order.

Tests should eventually support explicitly controlled:

• event order
• event timestamps
• watermarks
• late events

12. Flink Backend

Only after the IR + simulator work should Flink be introduced.

Streaming IR
     │
     ▼
Flink Backend
     │
     ▼
Flink SQL / executable representation

Flink owns:

• distributed execution
• checkpointing
• network shuffle
• recovery
• state backend
• backpressure
• parallelism

Streamform owns:

• semantic interpretation
• validation
• plan structure
• changelog meaning
• state compatibility
• developer workflow

13. Multi-Model DAG

Streamform eventually supports:

source orders
      ↓
clean_orders
      ↓
customer_metrics
      ↓
sink

Streamform’s own authoring syntax can initially provide:

source(...)
ref(...)

These do not need to depend on dbt. As built: a project is one StreamingPlan; every node knows its model; ref('m') is an edge from a model’s first operator to m’s Sink node, so a model is planned once no matter how many read it, and changelog mode crosses the boundary (see docs/project-format.md).

Later the dbt frontend can map dbt’s:

source()
ref()

into the same model.

14. Lifecycle / Reconciler

This should become one of the project’s most important systems.

Commands:

streamform plan
streamform apply
streamform inspect
streamform drain

rather than thinking primarily in terms of:

run
deploy

because a streaming application is persistent.

Conceptually:

Desired Plan
      │
      ▼
   compare
      ▲
      │
Running Plan

Output:

SAFE

STATE MIGRATION

STATE INCOMPATIBLE

BACKFILL REQUIRED

15. Native Rust Runtime

The simulator evolves gradually into a runtime.

Progression:

deterministic simulator
        ↓
Arrow batches
        ↓
async sources
        ↓
bounded channels
        ↓
backpressure
        ↓
persistent keyed state
        ↓
checkpointing
        ↓
recovery
        ↓
parallel operators
        ↓
distributed workers

Do not begin with distributed execution.

16. Queryable State

When persistent keyed state is implemented, design it so it can eventually be accessed externally.

Conceptually:

Runtime
   │
   ├── update state
   │
   └── query state
           │
           ▼
     customer_id=42

This makes future serving much easier.

17. API Serving

Later:

Kafka
 ↓
Streamform computation
 ↓
queryable keyed state
 ↓
HTTP / gRPC / Arrow Flight

The serving layer should be an extension of maintained state, rather than a separate duplicated pipeline.

18. Python

Python should be optional and delayed.

Potential future use:

pip install streamform

could provide:

from streamform import Project

project = Project.load(".")
project.test()

Possible implementation:

Rust core
   ↓
PyO3
   ↓
Python package

But only add this when:

• notebook integration matters
• Python programmatic APIs matter
• dbt ecosystem integration requires it

The CLI and compiler should remain Rust-native.

Streamform — Product roadmap

Status assessed on 2026-09-16 against the local checkout at 6e9e818. This is a source and documentation assessment, not a fresh certification of tests, releases, or external repositories.

This roadmap replaces the original phase-by-phase build order. The vision remains the product thesis; architecture sets the compiler boundaries; the implementation strategy (docs/implementation-strategy.md in the source repository) defines delivery work and acceptance gates. Future behavior described here is planned, not an existing CLI or file-format contract.

Product direction

Streamform should make a stateful streaming application understandable, testable, and safe to change. The initial user is an engineer who knows SQL, Kafka, Git, and CI, but should not need to become a Flink specialist to answer: “What will this application emit, and what happens to its state when I deploy this change?”

The near-term product is a Rust compiler, deterministic simulator, and deployment workflow over Flink. Its differentiator is the connection between tested semantics, explainable change impact, and an observed deployment. Native execution, serving, and a hosted control plane remain options after this workflow earns adoption.

The next product proof is a customer-metrics application that can be tested with late events and windows, deployed once, inspected, and restarted from verified state without duplicate jobs or unexplained loss. A changed grouping key must produce an actionable refusal or explicit rebuild plan, not an implied safe upgrade.

Where the project actually stands

The workspace version remains 0.4.0. Event time is already built; the immediate work is validating and packaging it, not implementing it again. The earlier roadmap’s Phase 0–6 work is largely present, Phase 7 is implemented but unreleased, and Phase 8 begins with tumbling windows.

Delivery order

Milestones are ordered by dependencies and evidence, not dates. v0.5 is the existing release target; later version numbers should be assigned when their gates are met.

Execute M0 first, then M1. Start the bounded M2/M3 feasibility investigations during M1 if capacity permits; they need not delay the window contract, and they must finish before upgrade promises are made. After M1, deliver M2 then M3 before adding sliding/session windows or starting a native runtime. This is work sequencing, not a requirement for parallel agents.

M0 — Release the capabilities already built

Close the v0.5 release around event time and the published protocol. Run the repository checks and real-cluster suite, preserve the documented processing-time retention limitation, and publish reproducible examples with the binary. A public quick start must work without cloning this private source repository.

Audit the release-facing prose: the README and introduction still say event time is absent; the Unreleased changelog contains an earlier intermediate claim that watermark fixtures are refused; contribution/release instructions and ecosystem availability statements are inconsistent with newer distribution choices. Classify external components as verified available, planned, or unavailable. Do not make the release depend on implementing every ecosystem promise; implement or explicitly defer each one.

Exit: a new user on a supported platform installs the release, gets the documented check/explain/test output from downloadable examples, validates the JSON reports against shipped schemas, and can follow the Flink guide with accurate caveats. Release tags and publication remain explicit release actions.

M1 — One complete window capability

Deliver event-time tumbling aggregation with final emission, append input, and fixed positive millisecond durations. Start with a single event-time lineage and existing aggregates. Keep window boundaries, emission, cleanup, and late-data semantics in the generic IR. The window proposal (docs/windows.md in the source repository) supersedes the previous draft and identifies the decisions the feasibility work must settle.

Do not ship a simulator-only interpretation while presenting it as portable to Flink. Reject unsupported combinations before submission. Add a canonical window example, versioned schema changes where needed, stable fingerprints, meaningful plan-diff reasons, and a differential integration test.

Exit: an engineer can predict when a window becomes final, observe no premature output, test boundary and late events without Docker, and run the supported equivalent on Flink. Memory is reclaimed when windows close; fixture exhaustion is not an implicit infinite watermark.

M2 — Make deployment observable and repeatable

Separate desired intent, last submitted deployment, and observed cluster state. A local applied record is evidence of a prior submission, not proof that a job is alive or that its state is restorable.

Add application/target ownership, read-only inspection, an immutable artifact identity, deployment attempts with recoverable state, and a single-writer guard. Repeating the same apply against the same verified running deployment should be a no-op. A timeout after submission is an unknown outcome requiring reconciliation, not permission to start another job. Replacements and rebuilds need explicit modes and sink consequences.

Exit: repeated apply, process interruption, missing jobs, stale records, and ambiguous HTTP outcomes all have tested outcomes. The CLI never announces a healthy deployment solely because a statement was accepted. Define remote gateway TLS, authentication, credential redaction, and supported deployment environments before claiming remote production readiness.

M3 — Turn plan advice into a supported upgrade workflow

First prove stop/savepoint/restore of the identical deployed artifact on the pinned Flink version, including source offsets and externally visible sink behavior. Then enable only the change classes for which the backend can prove physical state compatibility.

Streamform node ids and SHA-256 semantic fingerprints are necessary, but do not identify Flink’s physical operators or serializers. Investigate compiled plans and backend operator identity before selecting the deployment representation. If the current SQL Gateway approach cannot preserve identity, record that constraint and keep unsupported upgrade execution blocked; do not silently rebuild or create a new runtime to avoid the decision.

Keep semantic verdict and executable deployment action separate. A COMPATIBLE semantic change may still have no supported restore path; a STATE MIGRATION REQUIRED result does not mean a migrator exists. Rebuilds require a replayable source range and an explicit sink replacement/cutover policy.

Exit: automated tests demonstrate one supported state-preserving operation, unsupported operations fail before mutation, and partial failures leave sufficient evidence for operator recovery. Do not promise universal rollback or exactly-once delivery from a savepoint alone.

M4 — Continuous contracts

Start with a small set such as not-null and simple predicates, and one deterministic failure policy per release. Define contract placement relative to filters, windows, aggregation, and sink writes. Failure reports should identify the model, contract, and count without unexpectedly exporting event payloads.

Quarantine/dead-letter output is a separate sink with its own schema and delivery semantics. Add it only after its failure and retry behavior is specified. A runtime fail action needs a defined failed-job and restart policy.

Exit: fixture failures and runtime failures are observable and explainable, with a backend capability check for every supported action.

M5 — Expand from observed workloads

Prioritize requests by concrete applications and support cost. The candidate order is retraction-aware aggregation and explicit upsert/CDC inputs; multiple-source fixtures and watermark coordination; bounded interval or temporal joins; then additional windows. This order is provisional and may change with user evidence.

Aggregation over updates needs correct before/after and delete handling, including non-invertible aggregates. Multiple inputs need source-qualified fixtures, watermark minimum/idle-source rules, and bounded state policies before joins. Sliding windows multiply retained state; session windows can merge keys and require new emission and compatibility rules.

Small SQL additions such as CASE, casts, and selected scalar functions may be delivered between milestones when backed by real queries and parity tests. Avoid expanding syntax without executable semantics. dbt import and Python clients belong here only when a real integration need justifies them.

M6 — Keep expensive options conditional

The existing free binary and open-format product direction is retained. External library, runner, Action, and Cloud plans are adoption/distribution work, not substitutes for semantic or lifecycle correctness. Their current availability was not verified in this review.

Measures of progress

Track time from installation to first passing fixture; whether users can explain their changelog and plan verdict; time to diagnose a failed deployment; and successful recoveries without duplicate jobs or lost state. Begin with three representative design-partner applications as a proposed validation cohort, not an adoption claim.

Engineering gates are concrete: unchanged manifests keep their digests where promised; supported backend cases have differential tests; repeated apply creates no additional owned job; restore tests continue accumulated totals rather than restarting them; unsupported behavior fails early and names the limitation.

Window state is bounded in event time, not absolutely bounded in memory: stalled watermarks and unbounded key cardinality still require visibility and limits. Operational resource measurements belong in acceptance evidence, not an unqualified “bounded state” claim.

Mapping from the founding roadmap

For issue preparation, use the stable work-package ids and dependency rules in the implementation strategy (docs/implementation-strategy.md). Create detailed issues for M0–M3 first; keep later work as gated epics until the preceding evidence exists.