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.