Table options

A table’s layout is written as clauses after its columns. Everything else goes in a WITH (name = value, …) clause, where a value is a quoted string, a number, or a list in parentheses: publish = (delta, iceberg). Names aren’t case sensitive.

Tables

CREATE TABLE t (…), then its layout clauses in any order, then WITH (…). Which kind of table you get depends on the key:

• Append table: no key. Every row is kept.
• Keyed table (upsert table): a PRIMARY KEY. A new row with an existing key replaces the old one.
• Merge table: a key, and MERGE on every other column. Rows with the same key combine.

Clause	For	Meaning
PRIMARY KEY (a, b)	keyed, merge	the key, as a column or table constraint
PARTITION BY col, PARTITION BY day(col)	all	keep one partition per file. The units are year, month, day and hour, and need a timestamp or date column
CLUSTER BY (col1, col2)	all	sort every file by these columns, so a filter on them skips most of a file; two or more follow a Hilbert curve
SEQUENCE BY col	keyed	the row with the greatest value of this column wins, not the last to arrive
TTL col + INTERVAL '1 hour'	keyed	a row whose timestamp or date column is older than that is gone
col TYPE MERGE sum	merge	how the column combines: sum, count, min or max

Option	Value	For	Meaning
publish	delta, iceberg or (delta, iceberg)	all	also write the table as Delta Lake or Iceberg, for other engines. Default: none, or what --publish says
retention	'7 days', '12 hours', '0 seconds'	all	how long the table’s past is kept: read as it was with t AT (…), and, dropped, brought back with UNDROP TABLE. Default: a day

An option name that isn’t in this table is refused when you CREATE the table, and the error names it. In CREATE TABLE t … AS SELECT …, the clauses and WITH come before AS.

Each clause is also an option, for tools that write options: partition_by = 'day(ts)', cluster_by = 'a, b', order_by = 'seen', ttl = 'seen:3600' (seconds) and merge = 'amount:sum, n:count'. Giving a clause and its option both is refused. Spark’s PARTITIONED BY (col) is PARTITION BY col.

Change a table’s layout later with ALTER TABLE t CLUSTER BY (a, b), ALTER TABLE t TTL seen + INTERVAL '2 hours' and ALTER TABLE t SEQUENCE BY seen, and its options with ALTER TABLE t SET (publish = iceberg, retention = '30 days'). Files written from then on follow them, and files written before keep their order until they are merged. PARTITION BY can’t change, because each file holds one partition, and MERGE can’t change either.

Append table

CREATE TABLE clicks (ts TIMESTAMP, user_id BIGINT, url VARCHAR)
PARTITION BY day(ts)
CLUSTER BY (user_id)
WITH (publish = (delta, iceberg));
INSERT INTO clicks VALUES (TIMESTAMP '2026-01-01 10:00:00', 1, '/a'), (TIMESTAMP '2026-01-02 10:00:00', 2, '/b');
SELECT count(*) AS n FROM clicks;
ALTER TABLE clicks CLUSTER BY (user_id, url);
ALTER TABLE clicks SET (publish = iceberg);

Keyed table

SEQUENCE BY lets rows arrive out of order: below, the row with the later seen stays, even though the older one is written last. TTL drops a row an hour after its seen.

CREATE TABLE sessions (user_id BIGINT PRIMARY KEY, seen TIMESTAMP, page VARCHAR)
SEQUENCE BY seen
TTL seen + INTERVAL '1 hour';
INSERT INTO sessions VALUES (1, now(), '/home');
INSERT INTO sessions VALUES (1, TIMESTAMP '2000-01-01 00:00:00', '/old');
SELECT user_id, page FROM sessions;

user_id  page
1        /home

A keyed table also has a hidden _deleted column, so DELETE works. SELECT * leaves it out.

Merge table

Every column that isn’t part of the key needs a function. sum and count both add: each row you write carries its own count, as count(*) does in a view’s GROUP BY.

CREATE TABLE totals (
  region  VARCHAR PRIMARY KEY,
  amount  DOUBLE MERGE sum,
  n       BIGINT MERGE count,
  biggest DOUBLE MERGE max
);
INSERT INTO totals VALUES ('eu', 5.0, 1, 5.0);
INSERT INTO totals VALUES ('eu', 3.0, 1, 3.0), ('us', 2.0, 1, 2.0);
SELECT * FROM totals ORDER BY region;

region  amount  n  biggest
eu      8.0     2  5.0
us      2.0     1  2.0

What Pondra refuses:

You write	Error
an option name the table doesn’t take, such as cluser_by	CREATE TABLE … WITH (cluser_by = …): tables take publish, cluster_by, partition_by, ttl, order_by, merge
MERGE without a key	MERGE needs a key: a merge table combines the rows of each key
a merge table with a non-key column that has no MERGE	a merge table combines every column but its key: s needs a merge function too
b DOUBLE MERGE avg	MERGE: sum, count, min or max
publish = parquet	publish formats: delta, iceberg
CLUSTER BY a column the table lacks	CLUSTER BY: columns of the table
PARTITION BY week(ts)	PARTITION BY: a column, or year/month/day/hour(a timestamp column)
TTL on a table with no key, or a column that isn’t a timestamp or date	TTL: a timestamp or date column of a keyed table
TTL ts + INTERVAL '1 month'	an interval in seconds, minutes, hours, days or weeks
SEQUENCE BY a key column	SEQUENCE BY: a column of the table, not its key
PARTITION BY and partition_by both	partition_by is given twice, as a clause and in WITH (…)
ALTER TABLE t SET (partition_by = …)	PARTITION BY can't change
retention = 'soon'	retention: a time such as '7 days', '12 hours' or '0 seconds'

Materialized views

CREATE MATERIALIZED VIEW v [WITH (…)] AS SELECT … [EMIT FINAL]. Without options, the view is kept current with every row its table takes in. EMIT FINAL keeps each group of a time bucket once it is over (windows). The other options make windows, sessions and joins of two streams the older way. An option the view doesn’t know is an error that lists the ones it does.

Option	Value	Default	Meaning
lateness	a time, such as '10 seconds'	'0 seconds'	EMIT FINAL: how far out of order a row may arrive and still count
idle	a time	none	EMIT FINAL: once no newer row has come for this long, the watermark moves on with the clock
emit	'final'		the same as EMIT FINAL at the statement’s end, for clients that build the statement
refresh	'incremental', 'by key' or 'full'	chosen	how the view is kept current: from each write’s rows, in its commit; by working out again the groups a commit touched, right after it (kept by key); or by running its query whole again (run whole). Unset, the first of these that can keep it
lag	a time, such as '1 minute'	none	a view kept by key or run whole runs at most once this long; one that calls now() runs this often (a minute unless given). Refused on an incremental view
window	column name		the view’s window-start column, made by date_bin(…) in GROUP BY; each window goes to {view}_final once, when it is final
size_secs	seconds	60	how long a window is
slide_secs	seconds	none	a new window every this many seconds: a divisor of size_secs, less than it. Make the view’s date_bin this long
lateness_secs	seconds	0	how far out of order a row may arrive and still count
session	column name		the source’s event-time column: the view groups each key’s rows into sessions
gap_secs	seconds	1800	a session ends when its key has no row for this long
join	'streams'		pair rows of two tables as they arrive on either side
time	'col' or 'col1, col2'		the event-time column both tables have, or one for each table
within_secs	seconds	none	pair a row only with rows this close to it in time; needs time
start	'earliest' or 'latest'	'earliest'	for a view over a Kafka topic: where to start reading
history	'col' or 'col1, col2'		keep every version of each key, with __start_at and __end_at when read (history per key)
sequence_by	column name		a history view’s order of versions: a time, or a number that grows
delete_when	a condition	none	a history view’s version that ends its key, not shown

Windows: EMIT FINAL

GROUP BY a bucket of the source’s event-time column that only grows with time (date_bin, date_trunc, CAST(ts AS DATE), floor(date_part('epoch', ts) / 60), …) and end the statement with EMIT FINAL. The view keeps each group once, when the watermark (the newest event time in the source, less lateness) has passed it. A bucket that comes back round (hour(ts)) is refused. A row for a group already kept is late: the view keeps what it had, and pondra.flows counts it (late_rows). GROUP BY key, SESSION(ts, INTERVAL '30 minutes') keeps sessions instead, as session and gap_secs below do.

CREATE TABLE readings (ts TIMESTAMP, sensor VARCHAR, temp DOUBLE);
CREATE MATERIALIZED VIEW per_minute WITH (lateness = '10 seconds') AS
  SELECT date_bin(INTERVAL '1 minute', ts) AS minute, sensor, sum(temp) AS total, count(*) AS n
  FROM readings GROUP BY 1, 2
  EMIT FINAL;
INSERT INTO readings VALUES
  (TIMESTAMP '2026-01-01 10:00:05', 'a', 20.0), (TIMESTAMP '2026-01-01 10:00:40', 'a', 22.0),
  (TIMESTAMP '2026-01-01 10:01:30', 'a', 30.0), (TIMESTAMP '2026-01-01 10:03:00', 'a', 25.0);

sleep 2   # the leader looks for groups that are over twice a second

SELECT minute, sensor, total, n FROM per_minute ORDER BY minute;

minute               sensor  total  n
2026-01-01T10:00:00  a       42.0   2
2026-01-01T10:01:00  a       30.0   1

The 10:03 row moved the watermark to 10:02:50: the minutes before 10:02 are over. SHOW CREATE MATERIALIZED VIEW per_minute gives the statement back, EMIT FINAL and its options with it.

Windows and sessions as options

Before EMIT FINAL, windows were options, and they still work. A window closes when the watermark passes its end. The watermark is the newest event time in the source, less lateness_secs. The view keeps every window, open or not; each one goes once to {view}_final. A row that arrives after its window closed still updates the view, but not what was emitted.

Tumbling and sliding windows

CREATE TABLE gauges (ts TIMESTAMP, sensor VARCHAR, temp DOUBLE);
CREATE MATERIALIZED VIEW per_minute_open WITH (window = 'w', size_secs = 60, lateness_secs = 10) AS
  SELECT date_bin(INTERVAL '1 minute', ts) AS w, sensor, sum(temp) AS total, count(*) AS n
  FROM gauges GROUP BY 1, 2;
INSERT INTO gauges VALUES
  (TIMESTAMP '2026-01-01 10:00:05', 'a', 20.0), (TIMESTAMP '2026-01-01 10:00:40', 'a', 22.0),
  (TIMESTAMP '2026-01-01 10:01:30', 'a', 30.0), (TIMESTAMP '2026-01-01 10:03:00', 'a', 25.0);
SELECT * FROM per_minute_open ORDER BY w;

The view has a row for every window, still open or not. per_minute_open_final has the ones that are final, a moment after the watermark passes them: here the first two, because the 10:03 row moved the watermark past them.

sleep 5   # windows close a moment after the watermark passes them

SELECT w, sensor, total, n FROM per_minute_open_final ORDER BY w;

w                    sensor  total  n
2026-01-01T10:00:00  a       42.0   2
2026-01-01T10:01:00  a       30.0   1

A window made with these options keeps sum, count, min and max. For avg, stddev or HAVING, write it as GROUP BY a time bucket … EMIT FINAL (see Streaming).

For sliding windows, add slide_secs, and make the view’s date_bin that long. This makes a five-minute window every minute, and each row is added once, not once per window:

CREATE TABLE visits (ts TIMESTAMP, visitor VARCHAR);
CREATE MATERIALIZED VIEW per_five WITH (window = 'w', size_secs = 300, slide_secs = 60) AS
  SELECT date_bin(INTERVAL '1 minute', ts) AS w, visitor, count(*) AS n
  FROM visits GROUP BY 1, 2;

Session windows

A session is one key’s rows with no gap of gap_secs between them. The view gets a row for each session, once, with session_start and session_end (the last row plus the gap). The view’s GROUP BY columns are the keys.

CREATE TABLE pageviews (visitor VARCHAR, ts TIMESTAMP, amount BIGINT);
CREATE MATERIALIZED VIEW visitor_sessions WITH (session = 'ts', gap_secs = 30, lateness_secs = 5) AS
  SELECT visitor, count(*) AS n, sum(amount) AS spent FROM pageviews GROUP BY visitor;
INSERT INTO pageviews VALUES
  ('a', TIMESTAMP '2026-01-01 10:00:00', 2), ('a', TIMESTAMP '2026-01-01 10:00:10', 3),
  ('a', TIMESTAMP '2026-01-01 10:05:00', 1), ('b', TIMESTAMP '2026-01-01 10:10:00', 1);

sleep 5   # sessions close a moment after the watermark passes them

SELECT visitor, n, spent, session_start, session_end FROM visitor_sessions ORDER BY visitor, session_start;

visitor  n  spent  session_start        session_end
a        2  5      2026-01-01T10:00:00  2026-01-01T10:00:40
a        1  1      2026-01-01T10:05:00  2026-01-01T10:05:30

The rows at 10:05 and 10:10 closed the sessions before them: b’s session is still open.

A row that falls inside a session already emitted is late, and is left out.

Stream joins

A row of either table pairs with the other table’s rows when it arrives, and with those that arrive after it. Each pair is made once. within_secs bounds how far apart in time two rows may be.

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 orders VALUES (1, 9.5, TIMESTAMP '2026-01-01 10:00:00');
INSERT INTO payments VALUES (1, 9.5, TIMESTAMP '2026-01-01 10:02:00');

sleep 5   # the leader pairs rows just after each commit

SELECT id, amount, paid FROM paid;

id  amount  paid
1   9.5     9.5

A topic as a table

A view over another Kafka cluster’s topic (a lake attached with TYPE kafka) is filled from the records as they arrive, once each, and survives restarts. It takes one option, start. The cluster needs a secret for its address. This example reads the node’s own Kafka port, as a cluster you could have anywhere.

CREATE TABLE menu (id BIGINT, name VARCHAR);
INSERT INTO menu VALUES (1, 'tea'), (2, 'cake');
CREATE SECRET local_kafka (TYPE kafka, SECURITY_PROTOCOL 'PLAINTEXT', SCOPE 'kafka://127.0.0.1:9092');
ATTACH 'kafka://127.0.0.1:9092' AS k (TYPE kafka);
CREATE MATERIALIZED VIEW menu_in WITH (start = 'earliest') AS
  SELECT json_get_int(value, 'id') AS id, json_get_str(value, 'name') AS name FROM k.menu;

sleep 10   # the feed reads the topic in the background

SELECT id, name FROM menu_in ORDER BY id;

id  name
1   tea
2   cake

Functions and procedures

CREATE FUNCTION f(…) RETURNS type LANGUAGE python WITH (…) AS $$ … $$. Procedures take packages, entry and timeout. The Postgres words STRICT, IMMUTABLE, STABLE and VOLATILE work too, in place of strict and volatility. A SQL function takes no options.

Option	Value	Default	Meaning
vectorized	true	false	call the function once per batch, with pyarrow arrays in and out, not once per row
strict	true	false	a NULL argument gives NULL without calling it
volatility	'immutable', 'stable' or 'volatile'	'volatile'	a query calling a volatile Python function is never answered from the result cache
packages	'requests, jinja2'	none	packages the code needs; each node installs them once
entry	function name	none	the body is a module, and this function of it runs
timeout	seconds, at most a week	60 for a function; none for a procedure	how long one call (a function: one batch) may take before its worker is stopped
cache	'10 minutes', '30 seconds'	none	reuse an answer for the same arguments this long, so an API or a model isn’t called again

vectorized, cache and strict are for functions only. A Python function says what it returns, in RETURNS type or RETURNS TABLE (…).

CREATE FUNCTION double_all(x BIGINT) RETURNS BIGINT LANGUAGE python WITH (vectorized = true) AS $$
import pyarrow.compute as pc
return pc.multiply(x, 2)
$$;

CREATE FUNCTION shout(t VARCHAR) RETURNS VARCHAR LANGUAGE python WITH (entry = 'loud', strict = true) AS $$
def loud(t):
    return t.upper() + '!'
$$;

CREATE FUNCTION quick(n BIGINT) RETURNS BIGINT LANGUAGE python WITH (timeout = 1) AS $$
import time
time.sleep(n)
return n
$$;

CREATE FUNCTION stamp() RETURNS DOUBLE LANGUAGE python WITH (cache = '1 minute') AS $$
import time
return time.time()
$$;

SELECT double_all(21) AS v, shout('tea') AS s, shout(NULL) AS n, quick(0) AS q, stamp() = stamp() AS same;

v   s     n     q  same
42  TEA!  NULL  0  true

quick(3) would stop at one second with it took longer than 1 s, its limit (the worker was stopped).

The same options over HTTP

POST /tables/{name} takes a table as JSON, with the same options as keys: columns (pairs of a name and an Arrow type), key (a list), merge (an object of column to function), publish and cluster_by (lists), partition_by, ttl and order_by (strings). POST /views/{name} takes the view’s SQL as the body and its options as query parameters. See HTTP.

curl -s -X POST localhost:8080/tables/points -d '{"columns": [["id", "Int64"], ["seen", "Timestamp(Microsecond, None)"], ["label", "Utf8"]], "key": ["id"], "order_by": "seen", "publish": ["iceberg"]}'
curl -s -X POST 'localhost:8080/views/readings_per_hour?window=w&size_secs=3600' -d "SELECT date_bin(INTERVAL '1 hour', ts) AS w, sum(temp) AS total FROM readings GROUP BY 1"

From Python, db.view("name", "SELECT …", materialized=True, window="w", size_secs=60) sends the same options.