> ## Documentation Index
> Fetch the complete documentation index at: https://firebolt-aggregate-helm-docs-pr-30.mintlify.site/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 <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`             |

<Note>
  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.
</Note>

## 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

<Warning>
  **Private Preview Feature**

  This feature is currently in private preview. Contact [support@firebolt.io](mailto:support@firebolt.io) to request early access.
</Warning>

| 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](/guides/change-data-capture/postgres#read-raw-change-events) for delete and key-update row shapes.

### MongoDB streams

<Warning>
  **Private Preview Feature**

  This feature is currently in private preview. Contact [support@firebolt.io](mailto:support@firebolt.io) to request early access.
</Warning>

| 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)