SUBSCRIBE

SUBSCRIBE streams updates from a source, table, or view as they occur.

Conceptual framework

The SUBSCRIBE statement is a more general form of a SELECT statement. While a SELECT statement computes a relation at a moment in time, a subscribe operation computes how a relation changes over time.

Fundamentally, SUBSCRIBE produces a sequence of updates. An update describes either the insertion or deletion of a row to the relation at a specific time. Taken together, the updates describe the complete set of changes to a relation, in order, while SUBSCRIBE is active.

You can use SUBSCRIBE to:

• Power event processors that react to every change to a relation or an arbitrary SELECT statement.
• Replicate the complete history of a relation while SUBSCRIBE is active.

Syntax

WITH options

The following options are valid within the WITH clause.

Details

Output

SUBSCRIBE emits a sequence of updates as rows. Each row contains all of the columns of the subscribed relation or derived from the SELECT statement, prepended with several additional columns that describe the nature of the update:

AS OF

The AS OF clause allows specifying a timestamp at which the SUBSCRIBE should begin returning results, in order to inspect the historical state of a relation. If AS OF is specified, no rows whose timestamp is less than the specified timestamp will be returned. If the timestamp specified is earlier than the earliest historical state retained by the source relations, an error will be signaled.

Currently, all user-defined sources and tables have a retention window of one second, so AS OF is of limited usefulness except when subscribing to queries over certain internal relations.

UP TO

The UP TO clause allows specifying a timestamp at which the SUBSCRIBE will cease running. If UP TO is specified, no rows whose timestamp is greater than or equal to the specified timestamp will be returned.

Interaction of AS OF and UP TO

The lower timestamp bound specified by AS OF is inclusive, whereas the upper bound specified by UP TO is exclusive. Thus, a SUBSCRIBE query whose AS OF is equal to its UP TO will terminate after returning zero rows.

A SUBSCRIBE whose UP TO is less than its “as of” timestamp (whether that timestamp was specified in an AS OF clause or chosen by the system) will signal an error.

Duration

SUBSCRIBE will continue to run until canceled, the session ends, the UP TO timestamp is reached, or all updates have been presented. The latter case typically occurs when tailing constant views (e.g. CREATE VIEW v AS SELECT 1).

SNAPSHOT

By default, SUBSCRIBE begins by emitting a snapshot of the subscribed relation, which consists of a series of updates describing the contents of the relation at its initial timestamp. After the snapshot, SUBSCRIBE emits further updates as they occur.

For updates in the snapshot, the mz_timestamp field will be fast-forwarded to the initial timestamp. For example, an insert that occurred before the SUBSCRIBE began would appear in the snapshot.

To see only updates after the initial timestamp, specify WITH (SNAPSHOT = false).

PROGRESS

Intuitively, progress messages communicate that no updates have occurred in a given time window. Without explicit progress messages, it is impossible to distinguish between a stall in Materialize and a legitimate period of no updates.

If the PROGRESS option is specified via WITH (PROGRESS), an additional mz_progressed column appears in the output. When the column is false, the rest of the row is a valid update. When the column is true the rest of the row is not a valid update and its content should be ignored; the row exists only to communicate that timestamps have advanced.

Not all timestamps that appear will have a corresponding row with mz_progressed set to true. For example, the following is a valid sequence of updates:

mz_timestamp | mz_progressed | mz_diff | column1
-------------|---------------|---------|--------------
1            | false         | 1       | data
2            | false         | 1       | more data
3            | false         | 1       | even more data
4            | true          | NULL    | NULL

Notice how Materialize did not emit explicit progress messages for timestamps 1 or 2. The receipt of the update at timestamp 2 implies that there are no more updates for timestamp 1, because timestamps are always presented in non-decreasing order. The receipt of the explicit progress message at timestamp 4 implies that there are no more updates for either timestamp 2 or 3—but that there may be more data arriving at timestamp 4.

Examples

SUBSCRIBE produces rows similar to a SELECT statement, except that SUBSCRIBE may never complete. Many drivers buffer all results until a query is complete, and so will never return. Below are the recommended ways to work around this.

Creating a counter load generator

As an example, we’ll create a counter load generator that emits a row every second:

CREATE SOURCE counter
  FROM LOAD GENERATOR COUNTER
  WITH (SIZE = '3xsmall');

Subscribing with FETCH

The recommended way to use SUBSCRIBE is with DECLARE and FETCH. These must be used within a transaction, with a single DECLARE per transaction. This allows you to limit the number of rows and the time window of your requests. Next, let’s subscribe to the counter load generator source that we’ve created above.

First, declare a SUBSCRIBE cursor:

BEGIN;
DECLARE c CURSOR FOR SUBSCRIBE (SELECT * FROM counter);

Then, use FETCH in a loop to retrieve each batch of results as soon as it’s ready:

FETCH ALL c;

That will retrieve all of the rows that are currently available. If there are no rows available, it will wait until there are some ready and return those. A timeout can be used to specify a window in which to wait for rows. This will return up to the specified count (or ALL) of rows that are ready within the timeout. To retrieve up to 100 rows that are available in at most the next 1s:

FETCH 100 c WITH (timeout='1s');

To retrieve all available rows available over the next 1s:

FETCH ALL c WITH (timeout='1s');

A 0s timeout can be used to return rows that are available now without waiting:

FETCH ALL c WITH (timeout='0s');

Using clients

If you want to use SUBSCRIBE from an interactive SQL session (e.g.psql), wrap the query in COPY:

COPY (SUBSCRIBE (SELECT * FROM counter)) TO STDOUT;

Mapping rows to their updates

After all the rows from the SNAPSHOT have been transmitted, the updates will be emitted as they occur. How can you map each row to its corresponding update?

If your row has a unique column key, it is possible to map the update to its corresponding origin row; if the key is unknown, you can use the output of hash(columns_values) instead.

In the example above, Column 1 acts as the column key that uniquely identifies the origin row the update refers to; in case this was unknown, hashing the values from Column 1 to Column N would identify the origin row.

Dropping the counter load generator source

When you’re done, you can drop the counter load generator source:

DROP SOURCE counter;