Streaming

Streaming in Pondra is SQL that keeps running. A materialized view is a query the node keeps up to date as rows arrive, in the same commit as the rows, so a read never lags behind the table. Windows and sessions are EMIT FINAL on the same statement, and stream joins and as-of joins its options. Use this page when you’d otherwise reach for a Flink job.

Keep a query up to date

Make a table with a few rows in it. A view can start from rows that are already there.

CREATE TABLE clicks (user VARCHAR, page VARCHAR, ts TIMESTAMP, amount BIGINT);
INSERT INTO clicks VALUES
  ('ann', 'home', '2026-09-29 10:00:05', 1),
  ('bob', 'home', '2026-09-29 10:00:40', 2),
  ('ann', 'cart', '2026-09-29 10:01:10', 3);

SQL

CREATE MATERIALIZED VIEW IF NOT EXISTS spend AS
  SELECT user, count(*) AS n, sum(amount) AS total FROM clicks GROUP BY user;

Python

import pondra
db = pondra.connect("http://localhost:8080")
db.view("spend", "SELECT user, count(*) AS n, sum(amount) AS total FROM clicks GROUP BY user", materialized=True)

JavaScript

import { connect } from "pondra";
const db = connect("http://localhost:8080");
await db.view("spend", "SELECT user, count(*) AS n, sum(amount) AS total FROM clicks GROUP BY user", { materialized: true });

Now write more rows. The view already has them when the write is acknowledged:

INSERT INTO clicks VALUES ('ann', 'checkout', '2026-09-29 10:02:00', 5);
SELECT * FROM spend ORDER BY user;

user  n  total
ann   3  9
bob   1  2

A view without GROUP BY (a filter, a projection, a join) appends its rows to its own table. A view with GROUP BY keeps one running row per group, and any number of nodes add to it at once. A view of a filter or a projection also follows an UPDATE or DELETE of its table: a row that changes is changed in the view, and a row that no longer matches leaves it (see Change rows).

A grouped view takes the aggregates you would write by hand: count, sum, avg, min, max, stddev and variance (and their _samp and _pop forms), bool_and and bool_or, any expression over them, HAVING, and ORDER BY … LIMIT. Its table keeps running totals as rows arrive (sums, counts, and a variance’s moments), and each read works the answers out from them, so they are as current as the rows:

CREATE MATERIALIZED VIEW IF NOT EXISTS habits AS
  SELECT user, count(*) AS n, avg(amount) AS mean, stddev_pop(amount) AS spread
  FROM clicks GROUP BY user HAVING count(*) > 1;
SELECT * FROM habits;

user  n  mean  spread
ann   3  3.0   1.632993161855452

Medians and distinct counts: kept by key

count(DISTINCT …), median, percentiles, string_agg and array_agg need every row of a group, not running totals. A view of them is kept by key: after each commit, the groups its rows touched are worked out again from the table, and only those. Nothing is held in memory, and a write costs a read of its own groups’ rows, not of the table.

CREATE MATERIALIZED VIEW IF NOT EXISTS reach AS
  SELECT page, count(DISTINCT user) AS users, median(amount) AS typical
  FROM clicks GROUP BY page;
SELECT * FROM reach ORDER BY page;

page      users  typical
cart      1      3.0
checkout  1      5.0
home      2      1.5

Such a view is current a moment after the commit, not in it: read the table when you need the answer as of your own write. WITH (lag = '1 minute') works it out at most once a minute, and WITH (refresh = 'by key') keeps any grouped view this way. It reads one table, with no join or subquery, and groups by columns it selects.

Anything else: run whole

Every other query is a materialized view too: a top N (ORDER BY … LIMIT), a window function, DISTINCT, a subquery over its own table, a median across a join, or a filter on now(). Its query runs whole again after a commit changes a table it reads, at most so often that its runs take a tenth of the time (a query of 50 ms is current within half a second; one of 10 s, within 100 s). A view that calls now(), or reads another lake or files, also runs every minute, or every lag.

CREATE MATERIALIZED VIEW IF NOT EXISTS biggest AS
  SELECT user, page, amount FROM clicks ORDER BY amount DESC LIMIT 2;
SELECT * FROM biggest ORDER BY amount DESC;

user  page      amount
ann   checkout  5
ann   cart      3

A row that comes out the same stays as it is, so the view’s change feed shows only what moved. pondra.flows says how each view is kept (refresh: incremental, by key or full), why (reason), and which tables it follows. WITH (refresh = 'full') runs any view this way; WITH (refresh = 'incremental') refuses a view the rows alone can’t keep.

Caution

A view that keeps a min or max, joins another table, or is a window, session or stream join can’t take a row back, so an UPDATE or DELETE on its source is refused while it exists. DROP MATERIALIZED VIEW name first, or change a copy of the table. To change a view’s query, CREATE OR REPLACE MATERIALIZED VIEW makes it again from the rows already there. A materialized view may follow another, in the same commit: see Flows.

From a frame

A frame becomes a view the same way:

from pondra import col
by_page = db.table("clicks").group_by("page").agg(pondra.len().alias("views"))
by_page.to_view("views_by_page", materialized=True)
db.sql("SELECT * FROM views_by_page ORDER BY page").rows()

Windows: each group once it’s over

A grouped view keeps running totals: a group changes with every row that arrives. Add EMIT FINAL and the view keeps each group only once it is over, when event time has moved past it. Group by a time bucket of an event-time column: date_bin, date_trunc, a date, or any expression that only grows as time does.

CREATE TABLE taps (user VARCHAR, ts TIMESTAMP);

SQL

CREATE MATERIALIZED VIEW per_minute WITH (lateness = '10 seconds') AS
  SELECT date_bin(INTERVAL '1 minute', ts) AS minute, user, count(*) AS n
  FROM taps GROUP BY 1, 2
  EMIT FINAL;

Python

db.view("per_minute",
        "SELECT date_bin(INTERVAL '1 minute', ts) AS minute, user, count(*) AS n FROM taps GROUP BY 1, 2",
        materialized=True, lateness="10 seconds", emit="final")

JavaScript

import { connect } from "pondra";
const db = connect("http://localhost:8080");
await db.view("per_minute",
  "SELECT date_bin(INTERVAL '1 minute', ts) AS minute, user, count(*) AS n FROM taps GROUP BY 1, 2",
  { materialized: true, lateness: "10 seconds", emit: "final" });

lateness is how far out of order rows may arrive. The watermark is the newest event time in the stream less that lateness, and a minute is over when the watermark passes its end. A row that arrives before then counts, however out of order it is.

INSERT INTO taps VALUES
  ('ann', '2026-09-29 10:00:05'), ('ann', '2026-09-29 10:00:35'), ('bob', '2026-09-29 10:00:50'),
  ('ann', '2026-09-29 10:01:20'), ('bob', '2026-09-29 10:01:40'),
  ('bob', '2026-09-29 10:02:30');

The leader looks for groups that are over twice a second, so give it a moment:

sleep 2

SELECT minute, user, n FROM per_minute ORDER BY minute, user;

minute               user  n
2026-09-29T10:00:00  ann   2
2026-09-29T10:00:00  bob   1
2026-09-29T10:01:00  ann   1
2026-09-29T10:01:00  bob   1

The 10:02 minute isn’t there yet. The newest row is 10:02:30, so the watermark is 10:02:20, before the minute’s end. The view’s query run on its own answers it as it is now, open minute and all:

SELECT date_bin(INTERVAL '1 minute', ts) AS minute, user, count(*) AS n
FROM taps GROUP BY 1, 2 ORDER BY 1, 2;

A view made over rows already in the table starts from them, and keeps each group of them that is over at once.

Late rows

A row that arrives after its group was kept is late. It is in the table, and the view keeps what it had. pondra.flows counts each view’s late rows:

INSERT INTO taps VALUES ('ann', '2026-09-29 10:00:59');

sleep 1

SELECT name, late_rows FROM pondra.flows WHERE name = 'per_minute';

name        late_rows
per_minute  1

If the count keeps growing, raise lateness: each group is then kept that much later.

When the stream goes quiet

The watermark moves with the rows, so the last group of a stream that stops waits for the next row. idle moves it on with the clock once no newer row has come for that long:

CREATE MATERIALIZED VIEW per_minute_now WITH (idle = '30 seconds') AS
  SELECT date_trunc('minute', ts) AS minute, count(*) AS n FROM taps GROUP BY 1
  EMIT FINAL;

A row that comes for a group kept that way is late.

Sliding windows

A total over the last five minutes, every minute, is a window frame over the minutes kept. Each row is counted once, in its minute, and the frame adds five of them up when it is read:

SELECT minute, user,
  sum(n) OVER (PARTITION BY user ORDER BY minute
               RANGE BETWEEN INTERVAL '4 minutes' PRECEDING AND CURRENT ROW) AS last_five
FROM per_minute ORDER BY minute, user;

Session windows

A session is one key’s rows with no gap longer than an interval between them. Group by the key and SESSION(ts, INTERVAL …), then add EMIT FINAL. Pondra adds session_start and session_end (the last row plus the gap) and keeps each session once, whole, when the watermark passes its end.

CREATE TABLE visits (user VARCHAR, ts TIMESTAMP, amount BIGINT);
CREATE MATERIALIZED VIEW sessions WITH (lateness = '5 seconds') AS
  SELECT user, count(*) AS n, sum(amount) AS spent FROM visits
  GROUP BY user, SESSION(ts, INTERVAL '30 seconds')
  EMIT FINAL;
INSERT INTO visits VALUES
  ('ann', '2026-09-29 10:00:00', 2), ('ann', '2026-09-29 10:00:10', 2), ('ann', '2026-09-29 10:00:20', 2),
  ('bob', '2026-09-29 10:00:00', 2), ('bob', '2026-09-29 10:00:25', 2),
  ('bob', '2026-09-29 10:02:00', 2),
  ('cy',  '2026-09-29 10:05:00', 2);

sleep 2

SELECT user, n, spent, session_start, session_end FROM sessions ORDER BY user, session_start;

user  n  spent  session_start        session_end
ann   3  6      2026-09-29T10:00:00  2026-09-29T10:00:50
bob   2  4      2026-09-29T10:00:00  2026-09-29T10:00:55
bob   1  2      2026-09-29T10:02:00  2026-09-29T10:02:30

Bob’s rows at 10:00:25 and 10:02:00 are more than 30 seconds apart, so they are two sessions. Cy’s row moved the watermark to 10:04:55, which closed everything before it. A row that falls inside a session already kept is late, and is left out. A row behind the watermark whose session isn’t over yet still counts.

Caution

A session view starts from the moment it is created; rows already in the table are not part of it.

Note

Before EMIT FINAL, windows and sessions were options: WITH (window = 'w', size_secs = 60, lateness_secs = 10) kept every window, open or not, in the view and each one once in <view>_final; WITH (session = 'ts', gap_secs = 30) is the SESSION(…) above. Both still work (table options).

Join two streams

join = 'streams' pairs a row of either table with the other table’s rows when it arrives and with those that arrive after it, each pair once, in whichever order they come. within_secs bounds how far apart in time a pair may be, so only that much of the other table is read.

CREATE TABLE orders (id BIGINT, amount DOUBLE, ts TIMESTAMP);
CREATE TABLE payments (order_id BIGINT, paid DOUBLE, ts TIMESTAMP);
CREATE MATERIALIZED VIEW paid WITH (join = 'streams', time = 'ts', within_secs = 600) AS
  SELECT o.id, o.amount, p.paid
  FROM orders o JOIN payments p
    ON o.id = p.order_id AND p.ts BETWEEN o.ts AND o.ts + INTERVAL '10 minutes';
INSERT INTO payments VALUES (1, 9.5, '2026-09-29 10:00:30');
INSERT INTO orders VALUES (1, 9.5, '2026-09-29 10:00:00'), (2, 4.0, '2026-09-29 10:00:10');
INSERT INTO payments VALUES (2, 4.0, '2026-09-29 10:03:00'), (3, 1.0, '2026-09-29 10:30:00');

sleep 2

SELECT * FROM paid ORDER BY id;

id  amount  paid
1   9.5     9.5
2   4.0     4.0

Payment 1 arrived before its order and still paired with it. Payment 3 has no order.

Caution

Stream joins run on the leader alone, and start from the moment the view is made.

Keep the newest row by event time

A keyed table with SEQUENCE BY keeps, for each key, the row with the latest event time, not the last row to arrive. This is Flink’s deduplication by event time.

CREATE TABLE latest (k BIGINT PRIMARY KEY, ts BIGINT, v VARCHAR) SEQUENCE BY ts;
INSERT INTO latest VALUES (1, 100, 'new');
INSERT INTO latest VALUES (1, 50, 'old');
SELECT * FROM latest;

k  ts   v
1  100  new

As-of joins

ASOF JOIN gives each row the other table’s row as it was at that moment: the latest quote at or before each trade. MATCH_CONDITION takes >=, >, <= or <; a row with no match gets NULLs.

CREATE TABLE quotes (sym VARCHAR, ts TIMESTAMP, price DOUBLE);
CREATE TABLE trades (id BIGINT, sym VARCHAR, ts TIMESTAMP, qty BIGINT);
INSERT INTO quotes VALUES
  ('ABC', '2026-09-29 09:30:00', 10.0), ('ABC', '2026-09-29 09:31:00', 11.0), ('ABC', '2026-09-29 09:32:00', 12.0);
INSERT INTO trades VALUES
  (1, 'ABC', '2026-09-29 09:30:30', 5), (2, 'ABC', '2026-09-29 09:31:45', 2), (3, 'XYZ', '2026-09-29 09:31:45', 2);
SELECT t.id, t.qty, q.price, t.qty * q.price AS value
FROM trades t ASOF JOIN quotes q MATCH_CONDITION (t.ts >= q.ts) ON t.sym = q.sym
ORDER BY t.id;

id  qty  price  value
1   5    10.0   50.0
2   2    11.0   22.0
3   2    NULL   NULL

Put the same query in a materialized view and each new trade is priced as of its own event time, even when it arrives after a newer quote:

CREATE MATERIALIZED VIEW priced AS
  SELECT t.id, t.qty, q.price, t.qty * q.price AS value
  FROM trades t ASOF JOIN quotes q MATCH_CONDITION (t.ts >= q.ts) ON t.sym = q.sym;

Caution

A quote that arrives after a trade of its time doesn’t reach that trade. A keyed table keeps only its latest row per key, so an as-of join against one gives the latest, not the row of that moment: keep a table’s history as rows for that.

Tasks: any streaming SQL

A view covers filters, joins and aggregates. A task runs any SQL over the new rows of a table whenever some commit, and appends the result to a target table, exactly once. The SQL may read the target too, which is how a task keeps state. Make the target in SQL, keyed or not, then make the task with HTTP:

CREATE TABLE totals (user VARCHAR PRIMARY KEY, total BIGINT, n BIGINT);

curl -s -X POST localhost:8080/tasks/running_totals -d '{
  "source": "clicks", "target": "totals", "key": ["user"],
  "sql": "SELECT c.user, coalesce(max(t.total), 0) + sum(c.amount) AS total, coalesce(max(t.n), 0) + count(*) AS n FROM clicks c LEFT JOIN totals t ON c.user = t.user GROUP BY c.user"
}'

The task runs over each commit’s new rows and writes one row per user to totals, so totals holds each user’s running total. Make a task before the rows it should see arrive. Add "shards": 4, "shard_by": "user" to spread the state over the nodes by the hash of a column.

Caution

A task’s target must have exactly the columns the SQL returns. If the target doesn’t exist, the task makes it from the query’s output. A task can’t be dropped yet, and it sees new rows only, so a table’s rows can’t be changed while a task follows it.

Follow new rows

watch streams a table’s new rows the moment they commit, including a keyed table’s upserts and deletes. changes adds every UPDATE and DELETE, each row with its _change_type.

Python

import threading
feed = db.watch("clicks")
threading.Timer(0.3, lambda: db.append("clicks", [{"user": "cy", "page": "home", "ts": "2026-09-29T10:03:00", "amount": 4}])).start()
print(next(feed))

JavaScript

import { connect } from "pondra";
const db = connect("http://localhost:8080");
const feed = db.watch("clicks");
setTimeout(() => db.append("clicks", [{ user: "dee", page: "home", ts: "2026-09-29T10:04:00", amount: 1 }]), 300);
console.log((await feed.next()).value);
await feed.return();

Shell

curl -N "localhost:8080/watch/clicks"                 # new rows, as they commit (NDJSON)
curl -N "localhost:8080/watch/clicks?changes=true"    # every change
curl -N "localhost:8080/watch/clicks?after=1234"      # replay from a position first

A replay with after goes back as far as the node keeps its log: start it with --changelog-secs 86400 to keep a day.

Live queries

A live query sends its answer now and again within milliseconds of each commit that changes it. It isn’t sent again when the answer comes out the same, and commits to other tables don’t wake it. Nothing runs on the node once the client goes away.

Python

live = db.live("SELECT count(*) AS n FROM clicks")
print(next(live))                       # the answer now
db.append("clicks", [{"user": "eve", "page": "cart", "ts": "2026-09-29T10:05:00", "amount": 2}])
print(next(live))                       # and again after the commit
live.close()

JavaScript

import { connect } from "pondra";
const db = connect("http://localhost:8080");
const live = db.live("SELECT count(*) AS n FROM clicks");
console.log((await live.next()).value);
await db.append("clicks", [{ user: "fay", page: "cart", ts: "2026-09-29T10:06:00", amount: 2 }]);
console.log((await live.next()).value);
await live.return();

In an app, loop over it: for rows in db.live(sql): redraw(rows), or for await (const rows of db.live(sql)) in JavaScript. The Postgres port can’t push an answer, so it has no live queries.

What streaming doesn’t do yet

• There is one watermark per source table, not per partition or node. A source that goes quiet holds it: its last sessions wait for more rows (a window view’s idle moves it on).
• Top-N by event time, timers and MATCH_RECOGNIZE don’t exist.
• A window view’s late rows are counted (pondra.flows’ late_rows), not kept apart; there’s no side output for them. A session view leaves its late rows out without counting them yet.