STREAM SELECT

Grammar of continuous queries — relations, joins and TUMBLE, HOP or SESSION windows, filters, aggregates, ordering and EMIT.

STREAM SELECT opens a continuous query. How results are delivered and what a stream means over time are covered in Streaming reads. The examples stream the transfers table.

Grammar

STREAM SELECT <outputs>
  FROM <relation> [<join> …]
     | TUMBLE(<relation>, <time>, INTERVAL <size>)
     | HOP(<relation>, <time>, INTERVAL <step>, INTERVAL <size>)
     | SESSION(<relation>, [<time>,] INTERVAL <gap> [, INTERVAL <max_duration>])
  [WHERE …]
  [GROUP BY …]
  [ORDER BY …] [LIMIT <n>]
  [EMIT [STREAM | CHANGELOG] <trigger>]
  [SETTINGS <name> = <value>, …]

<time>    := <column> | NOW() | NOW64()
<trigger> := AFTER WINDOW CLOSE [WITH DELAY INTERVAL <n> <unit>] [AND TIMEOUT INTERVAL <n> <unit>]
           | PERIODIC INTERVAL <n> <unit> [REPEAT]
           | ON UPDATE [WITH BATCH INTERVAL <n> <unit>]
           | PER EVENT

Filtered tail

STREAM SELECT token, sender, recipient, amount
FROM transfers
WHERE chain_id = 1 AND token = 'USDC';

The stream sends the current rows that match, then each new matching row.

Grouped aggregate

STREAM SELECT chain_id, token, SUM(amount) AS volume, COUNT(*) AS transfers
FROM transfers
GROUP BY chain_id, token;

Top N

STREAM SELECT token, SUM(amount) AS volume
FROM transfers
WHERE chain_id = 1
GROUP BY token
ORDER BY volume DESC
LIMIT 10;

The result is kept ordered as volumes change.

Windows

A windowed query outputs window_start and window_end with its aggregates.

Event time

STREAM SELECT window_start, window_end, token, COUNT() AS transfers
FROM TUMBLE(transfers, transferred_at, INTERVAL 1 MINUTE)
GROUP BY window_start, window_end, token
EMIT AFTER WINDOW CLOSE;

STREAM SELECT window_start, window_end, SUM(amount) AS volume
FROM HOP(transfers, transferred_at, INTERVAL 1 MINUTE, INTERVAL 1 HOUR)
EMIT AFTER WINDOW CLOSE;

Window state is charged to the stream's share of the node's continuous-state memory. An hour-wide window that hops every minute keeps 60 open windows per event, so size windows to the share.

Processing time

Pass NOW() or NOW64() instead of a time column:

STREAM SELECT window_start, window_end, COUNT() AS transfers
FROM SESSION(transfers, NOW64(), INTERVAL 30 SECONDS, INTERVAL 5 MINUTES);

EMIT

EMIT STREAM releases append-style output and EMIT CHANGELOG output that carries retractions. The trigger decides when results are released.

Trigger	Releases results
`AFTER WINDOW CLOSE [WITH DELAY …] [AND TIMEOUT …]`	when a window closes
`PERIODIC INTERVAL … [REPEAT]`	on a fixed interval
`ON UPDATE [WITH BATCH INTERVAL …]`	when the result changes, optionally batched
`PER EVENT`	for every event

STREAM SELECT token, COUNT(*) AS transfers
FROM TUMBLE(transfers, transferred_at, INTERVAL 5 MINUTES)
GROUP BY token
EMIT CHANGELOG AFTER WINDOW CLOSE
  WITH DELAY INTERVAL 2 SECONDS
  AND TIMEOUT INTERVAL 30 SECONDS;

STREAM SELECT token, COUNT(*) AS transfers
FROM transfers
GROUP BY token
EMIT PERIODIC INTERVAL 1 SECOND REPEAT;

EMIT belongs to STREAM SELECT; AFTER WINDOW CLOSE applies to window relations.

Joins, views and federated relations

A STREAM SELECT joins relations, streams a logical or materialized view, and streams the external relations of a federation source.

SETTINGS is the final clause of a STREAM SELECT. max_stream_duration caps the request's lifetime and subscriber_idle_ttl sets how long a shared arrangement lingers after its last subscriber leaves. See Settings.

STREAM SELECT

On this page