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 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.
MongoDB: the stream binds to a collection and tracks a change-stream position. See the MongoDB CDC guide.

Any stream can be read with READ_STREAM. Postgres and MongoDB streams are usually consumed by a CDC table, which applies the changes continuously. Stream metadata is visible in information_schema.streams.

Kafka streams

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 for how messages map to columns.
`TOPIC`	The Kafka topic to bind the stream to. Required.
`LOCATION`	A Kafka 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`.

For RAW_TEXT and BINARY stream types, define a single column named data with the matching type.

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

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

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.
`LOCATION`	A Postgres location. 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.

CREATE STREAM orders_changes (
    order_id    BIGINT,
    customer_id BIGINT,
    amount      NUMERIC(12, 2)
)
TABLE = 'public.orders'
LOCATION = 'pg_prod';

MongoDB streams

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

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.
`LOCATION`	A MongoDB location. 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.

CREATE STREAM orders_changes
LOCATION = 'mongo_prod'
COLLECTION = 'app.orders';

ALTER STREAM

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:

ALTER STREAM <stream_name> SET PARTITION <partition_id> OFFSET <offset>;

Add a column to a Kafka stream’s schema:

ALTER STREAM <stream_name> ADD COLUMN <column_name> <data_type> [NULL | NOT NULL];

DROP STREAM

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 consuming the stream first; without its stream, the table’s ingest worker cannot make progress and reports errors in information_schema.cdc_ingests.

READ_STREAM: read data from a stream.
CREATE CDC TABLE: apply a change stream continuously.
information_schema.streams: stream metadata and consumer offsets.
Change data capture: concepts and guides.

Overview

Performance and Observability

Security

Self-Managed

Managed service

Guides

SQL reference

Release notes

API reference

Legal

Kafka streams

Kafka stream types

Postgres streams

MongoDB streams

ALTER STREAM

DROP STREAM

​Kafka streams

​Kafka stream types

​Postgres streams

​MongoDB streams

​ALTER STREAM

​DROP STREAM

​Related

Kafka streams

Kafka stream types

Postgres streams

MongoDB streams

ALTER STREAM

DROP STREAM

Related