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 default), 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. 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 => <limit>]
    [, snapshot => <boolean>]
)

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`. With `false`, the position is unchanged and the same data can be re-read; for Postgres streams such a peek runs on an ephemeral copy of the replication slot, so it is isolated and repeatable. Consuming requires a write transaction.	`BOOLEAN`
`LIMIT`	Maximum rows to read: per Kafka partition, or per batch for CDC streams. Default `1000`. `0` reads without bound.	`INT`, `BIGINT`
`snapshot`	Postgres and MongoDB streams only. When `true`, returns a full snapshot of the source table or collection instead of the change feed, as the initial backfill of a CDC table does. Postgres snapshot rows are tagged `$op = 'R'`; MongoDB snapshot rows are synthetic inserts. Default `false`.	`BOOLEAN`

Setting consume to true (the default) 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.

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, `T` truncate, `R` snapshot 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.
`$source_partition`	`BIGINT`	Leaf-partition ordinal for partitioned sources; `0` otherwise.
`$stream_row_number`	`BIGINT`	1-based position of the row within this statement’s read, in delivery order.

Rows arrive in commit order; uncommitted transactions are never surfaced. See Postgres CDC for delete and key-update row shapes.

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, `R` replace, `D` delete, `T` collection drop, rename, or invalidation.
`$doc`	`JSON`	Exact post-image of the document; `NULL` for deletes.
`$before`	`JSON`	Exact pre-image; `NULL` for inserts.
`$update_desc`	`JSON`	Updated fields, removed fields, and truncated arrays, for updates.
`$key`	`JSON`	The document key.
`$resume_token`	`TEXT`	The change-stream position of this event.
`$cluster_time`	`BIGINT`	Cluster timestamp, packed into a 64-bit integer.
`$wall_time`	`TIMESTAMPTZ`	Wall-clock time of the operation.
`$ns`	`TEXT`	Namespace, as `database.collection`.
`$stream_row_number`	`BIGINT`	1-based position of the row within this statement’s read, in delivery order.

A first consuming read with no stored position starts at the current end of the change feed; it does not replay history.

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.

Read a snapshot of a CDC source

SELECT count(*)
FROM READ_STREAM(STREAM orders_changes, consume => false, snapshot => true);

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

Read a snapshot of a CDC source

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

​Read a snapshot of a CDC source

​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

Read a snapshot of a CDC source

Filter and transform while reading

Use Kafka pseudo-columns

Managing consumed positions

Related