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.