READ_STREAM

A table-valued function (TVF) that reads from a stream: messages from a Kafka topic, or change events from a Postgres table or MongoDB collection. READ_STREAM returns a table with the stream’s declared columns plus source-specific metadata pseudo-columns. A stream tracks a consumed position (Kafka offsets, a Postgres replication-slot position, or a MongoDB resume token). When consume is true, the position advances with the transaction that reads, so each read returns only new data. Position updates commit atomically with the data they produced, which makes ingestion through READ_STREAM exactly-once. To load the current contents of a Postgres table or MongoDB collection rather than its changes, use READ_SNAPSHOT. For continuous ingestion of Postgres or MongoDB change feeds, prefer a CDC table, which runs this machinery for you; direct READ_STREAM over CDC streams is most useful for inspection and custom pipelines.

Syntax

READ_STREAM (
    STREAM <stream_name>
    [, consume => <boolean>]
    [, limit_rows => <rows>]
    [, timeout_millis => <milliseconds>]
)

Parameters

Parameter	Description	Supported input types
`STREAM`	The stream object to read from.	Stream reference
`consume`	Whether to advance the stream’s consumed position. Default `true` for Kafka streams; required for Postgres and MongoDB streams. With `false`, the position is unchanged and the same data can be re-read. Consuming requires a write transaction.	`BOOLEAN`
`limit_rows`	Row budget for one read: per Kafka partition, or per read for CDC streams. Default `1000` for Kafka, 2,000,000 for Postgres, 100,000 for MongoDB. For Postgres and MongoDB the budget is soft: a read bundles whole source transactions and stops at the first transaction boundary at or past the budget, so it can overshoot by the remainder of one transaction. It does not limit how large a source transaction may be; a larger transaction is applied in full.	`INT`, `BIGINT`
`timeout_millis`	Postgres and MongoDB streams only; a Kafka stream rejects it. The read window in milliseconds: how long one read keeps reading before it stops at the next source transaction boundary and returns what it has. Default `2000`, at most `300000` (five minutes). A read also stops once `limit_rows` is met, whichever comes first. A source that goes quiet does not end the read early, so every call takes the full window unless `limit_rows` is met first, including a peek with `consume => false`. For a quick look, pass a small `timeout_millis` or `limit_rows`.	`INT`, `BIGINT`

Setting consume to true makes the query a write operation. Read-only queries, such as standalone SELECT statements outside of CREATE TABLE AS SELECT or INSERT INTO ... SELECT, cannot consume from a stream. Either set consume => false or use READ_STREAM inside a write statement.

Return type

The result has the stream’s declared columns plus pseudo-columns that depend on the source. Pseudo-columns are not included in SELECT *; reference them explicitly, double-quoted (for example SELECT "$op", * FROM ...).

Kafka streams

Pseudo-column	Type	Description
`$offset`	`BIGINT`	The Kafka offset of the message within its partition.
`$partition_id`	`BIGINT`	The Kafka partition the message was read from.
`$timestamp`	`TIMESTAMP`	The Kafka broker timestamp of the message.
`$op`	`TEXT`	Always `I`: a Kafka message carries no operation, so every row is an insert.
`$stream_row_number`	`BIGINT`	1-based position of the row within this statement’s read, in delivery order. Within a partition this follows offset order.
`$stream_read_seq`	`BIGINT`	Number of the consuming read that returned the row: the same on every row of one read, and increasing with every consuming read. Selecting it with `consume => false` raises an error.

How raw messages map to columns is set by the stream’s TYPE (JSON, CSV, RAW_TEXT, BINARY); see Kafka stream types.

Postgres streams

Private Preview FeatureThis feature is currently in private preview. Contact support@firebolt.io to request early access.

Pseudo-column	Type	Description
`$op`	`TEXT`	Operation: `I` insert, `U` update, `D` delete, `R` snapshot row. A source truncate fails the read rather than producing a row.
`$lsn`	`BIGINT`	WAL position of the change, packed into a 64-bit integer.
`$commit_lsn`	`BIGINT`	WAL position of the enclosing transaction’s commit.
`$commit_ts`	`TIMESTAMPTZ`	Commit timestamp of the enclosing transaction.
`$xid`	`BIGINT`	Source transaction ID.
`$stream_row_number`	`BIGINT`	1-based position of the row within this statement’s read, in delivery order.
`$stream_read_seq`	`BIGINT`	Number of the consuming read that returned the row: the same on every row of one read, and increasing with every consuming read. Selecting it with `consume => false` raises an error.

Rows arrive in commit order; uncommitted transactions are never surfaced. One read returns as many whole source transactions as the source has ready, up to limit_rows, so a busy source is caught up in one statement rather than one statement per transaction. The result is delivered as several row batches; row order within a batch is source order, and batches themselves can arrive out of order, because a read decodes them in parallel by default. Order by $stream_row_number where delivery order matters. See Postgres CDC for delete and key-update row shapes. Exactly-once holds when the statement consumes everything the read delivers. Do not put a LIMIT above a consuming read: a LIMIT that is satisfied before the read has drained ends it early. The rows already delivered are committed, but the stream’s position is not advanced: nothing is lost, and the next read delivers those rows again, so such a statement is at-least-once. Use limit_rows to bound a read instead; it stops at transaction boundaries and moves the position.

MongoDB streams

Private Preview FeatureThis feature is currently in private preview. Contact support@firebolt.io to request early access.

Pseudo-column	Type	Description
`$op`	`TEXT`	Operation: `I` insert, `U` update or replace, `D` delete. Snapshot rows are inserts. After a `READ_SNAPSHOT`, an insert the snapshot may already hold reads as `U`.
`$cluster_time`	`BIGINT`	Cluster timestamp of the change, packed into a 64-bit integer; `NULL` for snapshot rows.
`$wall_time`	`TIMESTAMPTZ`	Wall-clock time of the operation on the source; `NULL` for snapshot rows.
`$stream_row_number`	`BIGINT`	1-based position of the row within this statement’s read, in delivery order.
`$stream_read_seq`	`BIGINT`	Number of the consuming read that returned the row: the same on every row of one read, and increasing with every consuming read. Selecting it with `consume => false` raises an error.

Besides the pseudo-columns, a MongoDB stream always has the columns _id and doc; see CREATE STREAM. A first consuming read starts at the position CREATE STREAM recorded; it does not replay older history. A read returns changes in the order the source committed them, until timeout_millis has passed or it has limit_rows rows, and then stops at the next transaction boundary. A multi-document transaction is never split across reads. A read can run up to 500 ms past timeout_millis while it waits for new changes. A consuming read that returns no rows still moves the stored position forward, which keeps an idle stream’s position inside the oplog. A drop, rename or invalidation of the collection ends its change stream: the read returns the changes before it, and the next read fails. A statement reads a MongoDB stream at most once, counting READ_STREAM and READ_SNAPSHOT together and a WITH query once per reference, because each read stores the stream’s whole position. A view cannot read a MongoDB stream.

Examples

Peek at a stream without consuming

SELECT *
FROM READ_STREAM(
    STREAM my_stream,
    consume => false
)
LIMIT 10;

Inspect change events on a CDC stream

SELECT "$op", "$commit_ts", order_id, customer_id
FROM READ_STREAM(STREAM orders_changes, consume => false);

Ingest Kafka messages into a managed table

INSERT INTO orders
SELECT *
FROM READ_STREAM(STREAM my_stream);

The stream’s offsets advance with the insert, so running the statement again returns only messages that arrived after the previous read.

Filter and transform while reading

SELECT
    STRING_TO_ARRAY(data, ',')[1]::INTEGER AS order_id,
    STRING_TO_ARRAY(data, ',')[2] AS name,
    STRING_TO_ARRAY(data, ',')[3]::INTEGER AS age
FROM READ_STREAM(STREAM my_raw_stream, consume => false)
WHERE STRING_TO_ARRAY(data, ',')[3]::INTEGER > 25;

Use Kafka pseudo-columns

SELECT
    "$offset",
    "$partition_id",
    "$timestamp",
    order_id,
    name
FROM READ_STREAM(STREAM my_stream, consume => false)
ORDER BY "$offset";

Managing consumed positions

Kafka offsets are visible in information_schema.streams and can be repositioned with ALTER STREAM ... SET PARTITION ... OFFSET .... Postgres streams track their position in the source’s replication slot; MongoDB streams store a resume token. Neither is manually repositionable; see the Postgres CDC and MongoDB CDC operations sections.

Overview

Performance and Observability

Security

Self-Managed

Managed service

Guides

SQL reference

Release notes

API reference

Legal

Syntax

Parameters

Return type

Kafka streams

Postgres streams

MongoDB streams

Examples

Peek at a stream without consuming

Inspect change events on a CDC stream

Ingest Kafka messages into a managed table

Filter and transform while reading

Use Kafka pseudo-columns

Managing consumed positions

​Syntax

​Parameters

​Return type

​Kafka streams

​Postgres streams

​MongoDB streams

​Examples

​Peek at a stream without consuming

​Inspect change events on a CDC stream

​Ingest Kafka messages into a managed table

​Filter and transform while reading

​Use Kafka pseudo-columns

​Managing consumed positions

​Related

Syntax

Parameters

Return type

Kafka streams

Postgres streams

MongoDB streams

Examples

Peek at a stream without consuming

Inspect change events on a CDC stream

Ingest Kafka messages into a managed table

Filter and transform while reading

Use Kafka pseudo-columns

Managing consumed positions

Related