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.

Keyboard shortcuts

Streamform