> ## 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 and syntax for the CREATE STREAM command for Kafka, Postgres, and MongoDB sources.

# CREATE STREAM

Creates a stream object that binds to an external change or message feed and defines the schema of its rows. The source kind is determined by the [location](/reference-sql/commands/data-definition/create-location) the stream references:

* **Kafka**: the stream binds to a topic and tracks consumer offsets per partition.
* **Postgres**: the stream binds to a table and owns a logical replication slot on the source. See the [Postgres CDC guide](/guides/change-data-capture/postgres).
* **MongoDB**: the stream binds to a collection and tracks a change-stream position. See the [MongoDB CDC guide](/guides/change-data-capture/mongodb).

Any stream can be read with [`READ_STREAM`](/reference-sql/functions-reference/table-valued/read_stream). Postgres and MongoDB streams are usually consumed by a [CDC table](/reference-sql/commands/data-definition/create-cdc-table), which applies the changes continuously. Stream metadata is visible in [`information_schema.streams`](/reference-sql/information-schema/streams).

## Kafka streams

```sql theme={"theme":{"light":"css-variables","dark":"css-variables"}}
CREATE STREAM [IF NOT EXISTS] <stream_name> (
    <column_name> <data_type> [NULL | NOT NULL]
    [, <column_name> <data_type> [NULL | NOT NULL] ...]
)
TOPIC = '<topic_name>'
LOCATION = '<location_name>'
TYPE = '<stream_type>'
[DESCRIPTION = '<description>']
```

| Parameter                   | Description                                                                                                        |
| :-------------------------- | :----------------------------------------------------------------------------------------------------------------- |
| `IF NOT EXISTS`             | If a stream with `<stream_name>` already exists, the statement succeeds without creating a new stream.             |
| `<stream_name>`             | An identifier for the stream. Must be unique within the database.                                                  |
| `<column_name> <data_type>` | The stream schema. See [stream types](#kafka-stream-types) for how messages map to columns.                        |
| `TOPIC`                     | The Kafka topic to bind the stream to. Required.                                                                   |
| `LOCATION`                  | A Kafka [location](/reference-sql/commands/data-definition/create-location) with the connection details. Required. |
| `TYPE`                      | How raw messages are parsed: `JSON`, `CSV`, `RAW_TEXT`, or `BINARY`. Required.                                     |
| `DESCRIPTION`               | Optional text description.                                                                                         |

### Kafka stream types

| Type       | Behavior                                                                                                            |
| :--------- | :------------------------------------------------------------------------------------------------------------------ |
| `JSON`     | Each message is parsed as a JSON object. Columns are extracted by name from JSON keys; missing keys produce `NULL`. |
| `CSV`      | Each message is treated as a single CSV row, mapped positionally to the column definitions.                         |
| `RAW_TEXT` | Each message is returned as a single `TEXT` column named `data`.                                                    |
| `BINARY`   | Each message is returned as a single `BYTEA` column named `data`.                                                   |

<Note>
  For `RAW_TEXT` and `BINARY` stream types, define a single column named `data` with the matching type.
</Note>

```sql theme={"theme":{"light":"css-variables","dark":"css-variables"}}
CREATE LOCATION my_kafka WITH (
    SOURCE = KAFKA,
    BROKERS = 'broker1:9092,broker2:9092'
);

CREATE STREAM clickstream (
    user_id INTEGER,
    event_type TEXT,
    page_url TEXT,
    event_time TIMESTAMP
)
TOPIC = 'web_events'
LOCATION = 'my_kafka'
TYPE = 'JSON';
```

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

```sql theme={"theme":{"light":"css-variables","dark":"css-variables"}}
CREATE STREAM [IF NOT EXISTS] <stream_name> (
    <column_name> <data_type>
    [, <column_name> <data_type> ...]
)
TABLE = '<schema>.<table>'
[SUPPORT_TOASTED_VALUES = TRUE]
LOCATION = '<location_name>'
[DESCRIPTION = '<description>']
```

| Parameter                   | Description                                                                                                                                                                                                                                                                                                                                                                        |
| :-------------------------- | :--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `<column_name> <data_type>` | The source columns to capture, with their Firebolt types. Required; Postgres streams have no schema inference. Types must be primitive.                                                                                                                                                                                                                                            |
| `TABLE`                     | The source table, as `<schema>.<table>`. Required.                                                                                                                                                                                                                                                                                                                                 |
| `SUPPORT_TOASTED_VALUES`    | How updates that leave a large out-of-line (TOAST) value unchanged are handled. `FALSE` (default): such an update stops ingestion with an error instead of corrupting the mirror. `TRUE`: the source table must use `REPLICA IDENTITY FULL`, and the unchanged value is recovered from the old row image. See [TOAST columns](/guides/change-data-capture/postgres#toast-columns). |
| `LOCATION`                  | A Postgres [location](/reference-sql/commands/data-definition/create-location-postgres). Required.                                                                                                                                                                                                                                                                                 |
| `DESCRIPTION`               | Optional text description.                                                                                                                                                                                                                                                                                                                                                         |

`CREATE STREAM` connects to the source and creates the stream's logical replication slot. If the source table is partitioned, leaf partitions are discovered automatically and one slot is created per leaf, each requiring a dedicated publication; see [Partitioned source tables](/guides/change-data-capture/postgres#partitioned-source-tables).

```sql theme={"theme":{"light":"css-variables","dark":"css-variables"}}
CREATE STREAM orders_changes (
    order_id    BIGINT,
    customer_id BIGINT,
    amount      NUMERIC(12, 2)
)
TABLE = 'public.orders'
LOCATION = 'pg_prod';
```

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

```sql theme={"theme":{"light":"css-variables","dark":"css-variables"}}
CREATE STREAM [IF NOT EXISTS] <stream_name> [(
    <column_name> <data_type>
    [, <column_name> <data_type> ...]
)]
LOCATION = '<location_name>'
COLLECTION = '<database>.<collection>'
[KEY = '<column_name>']
[DESCRIPTION = '<description>']
```

| Parameter                   | Description                                                                                                                                                                                                        |
| :-------------------------- | :----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `<column_name> <data_type>` | Document fields to capture as typed columns. Optional: omit the list to infer columns from the collection's `$jsonSchema` validator. See [schema inference](/guides/change-data-capture/mongodb#schema-inference). |
| `LOCATION`                  | A MongoDB [location](/reference-sql/commands/data-definition/create-location-mongodb). Required.                                                                                                                   |
| `COLLECTION`                | The source collection, as `<database>.<collection>`. Required.                                                                                                                                                     |
| `KEY`                       | The column used as the merge key downstream. Optional; defaults to `'_id'`.                                                                                                                                        |
| `DESCRIPTION`               | Optional text description.                                                                                                                                                                                         |

```sql theme={"theme":{"light":"css-variables","dark":"css-variables"}}
CREATE STREAM orders_changes
LOCATION = 'mongo_prod'
COLLECTION = 'app.orders';
```

<h2 id="alter-stream">
  ALTER STREAM
</h2>

`ALTER STREAM` applies to Kafka streams only. For Postgres and MongoDB streams, schema changes are drop-and-recreate.

Reposition a Kafka stream's consumer offset on a partition:

```sql theme={"theme":{"light":"css-variables","dark":"css-variables"}}
ALTER STREAM <stream_name> SET PARTITION <partition_id> OFFSET <offset>;
```

Add a column to a Kafka stream's schema:

```sql theme={"theme":{"light":"css-variables","dark":"css-variables"}}
ALTER STREAM <stream_name> ADD COLUMN <column_name> <data_type> [NULL | NOT NULL];
```

## DROP STREAM

```sql theme={"theme":{"light":"css-variables","dark":"css-variables"}}
DROP STREAM [IF EXISTS] <stream_name>;
```

Removes the stream definition and its tracked position. Data already ingested into Firebolt tables is not affected. Source-side effects:

* **Kafka**: none; the topic is untouched.
* **Postgres**: the stream's replication slot (or slots, for a partitioned source) is dropped on the source, so it stops retaining WAL. If the source is unreachable, the stream is still dropped and the error message names the leftover slot for manual removal.
* **MongoDB**: none; change streams hold no server-side resource.

Drop or suspend any [CDC table](/reference-sql/commands/data-definition/create-cdc-table) consuming the stream first; without its stream, the table's ingest worker cannot make progress and reports errors in [`information_schema.cdc_ingests`](/reference-sql/information-schema/cdc-ingests).

## Related

* [`READ_STREAM`](/reference-sql/functions-reference/table-valued/read_stream): read data from a stream.
* [CREATE CDC TABLE](/reference-sql/commands/data-definition/create-cdc-table): apply a change stream continuously.
* [`information_schema.streams`](/reference-sql/information-schema/streams): stream metadata and consumer offsets.
* [Change data capture](/guides/change-data-capture): concepts and guides.