> ## Documentation Index > Fetch the complete documentation index at: https://docs.firebolt.io/llms.txt > Use this file to discover all available pages before exploring further. > Reference material for READ_STREAM function # READ_STREAM A table-valued function (TVF) that reads from a [stream](/reference-sql/commands/data-definition/create-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](/reference-sql/commands/data-definition/create-cdc-table), which runs this machinery for you; direct `READ_STREAM` over CDC streams is most useful for inspection and custom pipelines. ## Syntax ```sql theme={"theme":{"light":"css-variables","dark":"css-variables"}} READ_STREAM ( STREAM [, consume => ] [, limit_rows => ] [, snapshot => ] ) ``` ## 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_rows` | Row budget for one read: per Kafka partition, or per batch for CDC streams. Default `1000` for Kafka, 2,000,000 for Postgres. For Postgres 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` | | `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](/reference-sql/commands/data-definition/create-stream#kafka-stream-types). ### Postgres streams **Private Preview Feature** This feature is currently in private preview. Contact [support@firebolt.io](mailto: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. 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](/guides/change-data-capture/postgres#read-raw-change-events) 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 Feature** This feature is currently in private preview. Contact [support@firebolt.io](mailto: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 ```sql theme={"theme":{"light":"css-variables","dark":"css-variables"}} SELECT * FROM READ_STREAM( STREAM my_stream, consume => false ) LIMIT 10; ``` ### Inspect change events on a CDC stream ```sql theme={"theme":{"light":"css-variables","dark":"css-variables"}} SELECT "$op", "$commit_ts", order_id, customer_id FROM READ_STREAM(STREAM orders_changes, consume => false); ``` ### Ingest Kafka messages into a managed table ```sql theme={"theme":{"light":"css-variables","dark":"css-variables"}} 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 ```sql theme={"theme":{"light":"css-variables","dark":"css-variables"}} SELECT count(*) FROM READ_STREAM(STREAM orders_changes, consume => false, snapshot => true); ``` ### Filter and transform while reading ```sql theme={"theme":{"light":"css-variables","dark":"css-variables"}} 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 ```sql theme={"theme":{"light":"css-variables","dark":"css-variables"}} 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`](/reference-sql/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](/guides/change-data-capture/postgres#operations) and [MongoDB CDC](/guides/change-data-capture/mongodb#operations) operations sections. ## Related * [CREATE STREAM](/reference-sql/commands/data-definition/create-stream) * [CREATE CDC TABLE](/reference-sql/commands/data-definition/create-cdc-table) * [Change data capture](/guides/change-data-capture)