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

Mode	Meaning	Where it comes from
`Append`	Every row is a new, independent fact.	Sources; filters and projections over append input.
`Upsert(key)`	Rows are identified by a key; a later row for a key supersedes the earlier one.	`GROUP BY` (keyed by the grouping columns); a declared `upsert` model with a `key`.
`Retract`	Changes are emitted as retractions followed by additions.	Not produced by any operator yet; reserved.

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.