createStream

Last updated: January 27, 2026

Creates a new stream in Snowflake. Streams track data manipulation language (DML) changes made to tables, views, and other objects, enabling change data capture (CDC) workflows.

Note: Automatic rollback drops the stream.

Known limitation: This change type does not support database inspection features (snapshot, diff, diff-changelog, and generate-changelog commands).

Available attributes

Attribute	Type	Description	Required
`catalogName`	String	Name of the catalog (database) where the stream will be created	No
`schemaName`	String	Name of the schema where the stream will be created	No
`streamName`	String	Name of the stream to create	Yes
`sourceName`	String	Name of the source object (table, view, stage, etc.)	Yes
`sourceCatalogName`	String	Name of the catalog containing the source object	No
`sourceSchemaName`	String	Name of the schema containing the source object	No
`comment`	String	Comment describing the stream	No
`copyGrants`	Boolean	If true, copies grants from the source object	No
`replaceIfExists`	Boolean	If true, replaces the stream if it already exists using `OR REPLACE`*	No
`ifNotExists`	Boolean	If true, only creates if the stream doesn't exist using `IF NOT EXISTS`*	No
`tags`	Nested	Key-value pairs of tags to apply	No

*replaceIfExists and ifNotExists are mutually exclusive.

Stream type elements

Exactly one stream type must be specified:

Element	Description
`tableStream`	For tracking changes to standard tables
`viewStream`	For tracking changes to views
`externalTableStream`	For tracking changes to external tables
`stageStream`	For tracking directory table changes on stages
`dynamicTableStream`	For tracking changes to dynamic tables
`eventTableStream`	For tracking changes to event tables

tableStream / viewStream nested attributes

Attribute	Type	Description	Required
`appendOnly`	Boolean	If true, creates append-only stream	No
`showInitialRows`	Boolean	If true, includes initial data in stream	No
`timeTravelClause`	Nested	Time travel specification	No

externalTableStream nested attributes

Attribute	Type	Description	Required
`insertOnly`	Boolean	If true, only tracks inserts	No
`timeTravelClause`	Nested	Time travel specification	No

timeTravelClause nested attributes

Attribute	Type	Description	Required
`timeTravelType`	String	`AT` or `BEFORE`	Yes
`offset`	String	Offset value (e.g., `-1`)*	No
`timestamp`	String	Timestamp value*	No
`statement`	String	Statement ID*	No
`stream`	String	Stream name for reference*	No

*Exactly one of offset, timestamp, statement, or stream must be specified.

<?xml version="1.0" encoding="UTF-8"?>
<databaseChangeLog
        xmlns="http://www.liquibase.org/xml/ns/dbchangelog"
        xmlns:pro-snowflake="http://www.liquibase.org/xml/ns/pro-snowflake"
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:schemaLocation="http://www.liquibase.org/xml/ns/dbchangelog
        http://www.liquibase.org/xml/ns/dbchangelog/dbchangelog-latest.xsd
        http://www.liquibase.org/xml/ns/pro-snowflake
        http://www.liquibase.org/xml/ns/pro-snowflake/liquibase-pro-snowflake-latest.xsd">

    <!-- Table stream with time travel -->
    <changeSet id="create-table-stream" author="examples">
        <pro-snowflake:createStream streamName="TABLE_STREAM"
                                    sourceName="SOURCE_TABLE"
                                    replaceIfExists="true"
                                    copyGrants="true"
                                    comment="Stream for tracking table changes">
            <pro-snowflake:tableStream showInitialRows="true" appendOnly="true">
                <pro-snowflake:timeTravelClause timeTravelType="AT">
                    <pro-snowflake:offset>-1</pro-snowflake:offset>
                </pro-snowflake:timeTravelClause>
            </pro-snowflake:tableStream>
            <pro-snowflake:tags>
                <pro-snowflake:entry key="MYDB.PUBLIC.environment" value="production"/>
            </pro-snowflake:tags>
        </pro-snowflake:createStream>
    </changeSet>

    <!-- View stream -->
    <changeSet id="create-view-stream" author="examples">
        <pro-snowflake:createStream streamName="VIEW_STREAM"
                                    sourceName="SOURCE_VIEW"
                                    ifNotExists="true">
            <pro-snowflake:viewStream appendOnly="true"/>
        </pro-snowflake:createStream>
    </changeSet>

    <!-- Stage stream -->
    <changeSet id="create-stage-stream" author="examples">
        <pro-snowflake:createStream streamName="STAGE_STREAM"
                                    sourceName="DIRECTORY_STAGE"
                                    comment="Stream for directory table">
            <pro-snowflake:stageStream/>
        </pro-snowflake:createStream>
    </changeSet>

    <!-- External table stream with tags -->
    <changeSet id="create-external-table-stream" author="examples">
        <pro-snowflake:createStream streamName="EXTERNAL_TABLE_STREAM"
                                    sourceName="SOURCE_EXTERNAL_TABLE"
                                    comment="External table stream">
            <pro-snowflake:externalTableStream insertOnly="true"/>
            <pro-snowflake:tags>
                <pro-snowflake:entry key="MYDB.PUBLIC.environment" value="prod"/>
            </pro-snowflake:tags>
        </pro-snowflake:createStream>
    </changeSet>

    <!-- Dynamic table stream -->
    <changeSet id="create-dynamic-table-stream" author="examples">
        <pro-snowflake:createStream streamName="DYNAMIC_TABLE_STREAM"
                                    sourceName="SOURCE_DYNAMIC_TABLE">
            <pro-snowflake:dynamicTableStream/>
        </pro-snowflake:createStream>
    </changeSet>

    <!-- Event table stream -->
    <changeSet id="create-event-table-stream" author="examples">
        <pro-snowflake:createStream streamName="EVENT_TABLE_STREAM"
                                    sourceName="SOURCE_EVENT_TABLE">
            <pro-snowflake:eventTableStream/>
        </pro-snowflake:createStream>
    </changeSet>

</databaseChangeLog>

{
  "databaseChangeLog": [
    {
      "changeSet": {
        "id": "create-table-stream",
        "author": "examples",
        "changes": [
          {
            "createStream": {
              "streamName": "TABLE_STREAM",
              "sourceName": "SOURCE_TABLE",
              "replaceIfExists": true,
              "copyGrants": true,
              "comment": "Stream for tracking table changes",
              "tableStream": {
                "showInitialRows": true,
                "appendOnly": true,
                "timeTravelClause": {
                  "timeTravelType": "AT",
                  "offset": "-1"
                }
              },
              "tags": [
                {
                  "entry": {
                    "key": "MYDB.PUBLIC.environment",
                    "value": "production"
                  }
                }
              ]
            }
          }
        ]
      }
    },
    {
      "changeSet": {
        "id": "create-view-stream",
        "author": "examples",
        "changes": [
          {
            "createStream": {
              "streamName": "VIEW_STREAM",
              "sourceName": "SOURCE_VIEW",
              "ifNotExists": true,
              "viewStream": {
                "appendOnly": true
              }
            }
          }
        ]
      }
    },
    {
      "changeSet": {
        "id": "create-stage-stream",
        "author": "examples",
        "changes": [
          {
            "createStream": {
              "streamName": "STAGE_STREAM",
              "sourceName": "DIRECTORY_STAGE",
              "comment": "Stream for directory table",
              "stageStream": {}
            }
          }
        ]
      }
    },
    {
      "changeSet": {
        "id": "create-external-table-stream",
        "author": "examples",
        "changes": [
          {
            "createStream": {
              "streamName": "EXTERNAL_TABLE_STREAM",
              "sourceName": "SOURCE_EXTERNAL_TABLE",
              "comment": "External table stream",
              "externalTableStream": {
                "insertOnly": true
              },
              "tags": [
                {
                  "entry": {
                    "key": "MYDB.PUBLIC.environment",
                    "value": "prod"
                  }
                }
              ]
            }
          }
        ]
      }
    },
    {
      "changeSet": {
        "id": "create-dynamic-table-stream",
        "author": "examples",
        "changes": [
          {
            "createStream": {
              "streamName": "DYNAMIC_TABLE_STREAM",
              "sourceName": "SOURCE_DYNAMIC_TABLE",
              "dynamicTableStream": {}
            }
          }
        ]
      }
    },
    {
      "changeSet": {
        "id": "create-event-table-stream",
        "author": "examples",
        "changes": [
          {
            "createStream": {
              "streamName": "EVENT_TABLE_STREAM",
              "sourceName": "SOURCE_EVENT_TABLE",
              "eventTableStream": {}
            }
          }
        ]
      }
    }
  ]
}

databaseChangeLog:
  #Table stream with time travel
  - changeSet:
      id: create-table-stream
      author: examples
      changes:
        - createStream:
            streamName: TABLE_STREAM
            sourceName: SOURCE_TABLE
            replaceIfExists: true
            copyGrants: true
            comment: Stream for tracking table changes
            tableStream:
              showInitialRows: true
              appendOnly: true
              timeTravelClause:
                timeTravelType: AT
                offset: "-1"
            tags:
              - entry:
                  key: MYDB.PUBLIC.environment
                  value: production

  #View stream
  - changeSet:
      id: create-view-stream
      author: examples
      changes:
        - createStream:
            streamName: VIEW_STREAM
            sourceName: SOURCE_VIEW
            ifNotExists: true
            viewStream:
              appendOnly: true

  #Stage stream
  - changeSet:
      id: create-stage-stream
      author: examples
      changes:
        - createStream:
            streamName: STAGE_STREAM
            sourceName: DIRECTORY_STAGE
            comment: Stream for directory table
            stageStream: {}

  #External table stream with tags
  - changeSet:
      id: create-external-table-stream
      author: examples
      changes:
        - createStream:
            streamName: EXTERNAL_TABLE_STREAM
            sourceName: SOURCE_EXTERNAL_TABLE
            comment: External table stream
            externalTableStream:
              insertOnly: true
            tags:
              - entry:
                  key: MYDB.PUBLIC.environment
                  value: prod

  #Dynamic table stream
  - changeSet:
      id: create-dynamic-table-stream
      author: examples
      changes:
        - createStream:
            streamName: DYNAMIC_TABLE_STREAM
            sourceName: SOURCE_DYNAMIC_TABLE
            dynamicTableStream: {}

  #Event table stream
  - changeSet:
      id: create-event-table-stream
      author: examples
      changes:
        - createStream:
            streamName: EVENT_TABLE_STREAM
            sourceName: SOURCE_EVENT_TABLE
            eventTableStream: {}

--liquibase formatted sql

--changeset examples:table-stream-with-time-travel-and-tags
-- Table stream with time travel and tags
CREATE OR REPLACE STREAM MYDB.PUBLIC.TABLE_STREAM
WITH TAG (MYDB.PUBLIC.environment = 'production')
COPY GRANTS
ON TABLE MYDB.PUBLIC.SOURCE_TABLE
AT (OFFSET => -1)
APPEND_ONLY = TRUE
SHOW_INITIAL_ROWS = TRUE
COMMENT = 'Stream for tracking table changes';

--changeset examples:view-stream
-- View stream
CREATE STREAM IF NOT EXISTS MYDB.PUBLIC.VIEW_STREAM
ON VIEW MYDB.PUBLIC.SOURCE_VIEW
APPEND_ONLY = TRUE;

--changeset examples:stage-stream
-- Stage stream
CREATE STREAM MYDB.PUBLIC.STAGE_STREAM
ON STAGE MYDB.PUBLIC.DIRECTORY_STAGE
COMMENT = 'Stream for directory table';

--changeset examples:external-table-stream-with-tags
-- External table stream with tags
CREATE STREAM MYDB.PUBLIC.EXTERNAL_TABLE_STREAM
WITH TAG (MYDB.PUBLIC.environment = 'prod')
ON EXTERNAL TABLE MYDB.PUBLIC.SOURCE_EXTERNAL_TABLE
INSERT_ONLY = TRUE
COMMENT = 'External table stream';

--changeset examples:dynamic-table-stream
-- Dynamic table stream
CREATE STREAM MYDB.PUBLIC.DYNAMIC_TABLE_STREAM
ON DYNAMIC TABLE MYDB.PUBLIC.SOURCE_DYNAMIC_TABLE;

--changeset examples:event-table-stream
-- Event table stream
CREATE STREAM MYDB.PUBLIC.EVENT_TABLE_STREAM
ON EVENT TABLE MYDB.PUBLIC.SOURCE_EVENT_TABLE;

<?xml version="1.0" encoding="UTF-8"?>
<databaseChangeLog
        xmlns="http://www.liquibase.org/xml/ns/dbchangelog"
        xmlns:pro-snowflake="http://www.liquibase.org/xml/ns/pro-snowflake"
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:schemaLocation="http://www.liquibase.org/xml/ns/dbchangelog
        http://www.liquibase.org/xml/ns/dbchangelog/dbchangelog-latest.xsd
        http://www.liquibase.org/xml/ns/pro-snowflake
        http://www.liquibase.org/xml/ns/pro-snowflake/liquibase-pro-snowflake-latest.xsd">

    <!-- Table stream with time travel -->
    <changeSet id="create-table-stream" author="examples">
        <pro-snowflake:createStream streamName="TABLE_STREAM"
                                    sourceName="SOURCE_TABLE"
                                    replaceIfExists="true"
                                    copyGrants="true"
                                    comment="Stream for tracking table changes">
            <pro-snowflake:tableStream showInitialRows="true" appendOnly="true">
                <pro-snowflake:timeTravelClause timeTravelType="AT">
                    <pro-snowflake:offset>-1</pro-snowflake:offset>
                </pro-snowflake:timeTravelClause>
            </pro-snowflake:tableStream>
            <pro-snowflake:tags>
                <pro-snowflake:entry key="MYDB.PUBLIC.environment" value="production"/>
            </pro-snowflake:tags>
        </pro-snowflake:createStream>
    </changeSet>

    <!-- View stream -->
    <changeSet id="create-view-stream" author="examples">
        <pro-snowflake:createStream streamName="VIEW_STREAM"
                                    sourceName="SOURCE_VIEW"
                                    ifNotExists="true">
            <pro-snowflake:viewStream appendOnly="true"/>
        </pro-snowflake:createStream>
    </changeSet>

    <!-- Stage stream -->
    <changeSet id="create-stage-stream" author="examples">
        <pro-snowflake:createStream streamName="STAGE_STREAM"
                                    sourceName="DIRECTORY_STAGE"
                                    comment="Stream for directory table">
            <pro-snowflake:stageStream/>
        </pro-snowflake:createStream>
    </changeSet>

    <!-- External table stream with tags -->
    <changeSet id="create-external-table-stream" author="examples">
        <pro-snowflake:createStream streamName="EXTERNAL_TABLE_STREAM"
                                    sourceName="SOURCE_EXTERNAL_TABLE"
                                    comment="External table stream">
            <pro-snowflake:externalTableStream insertOnly="true"/>
            <pro-snowflake:tags>
                <pro-snowflake:entry key="MYDB.PUBLIC.environment" value="prod"/>
            </pro-snowflake:tags>
        </pro-snowflake:createStream>
    </changeSet>

    <!-- Dynamic table stream -->
    <changeSet id="create-dynamic-table-stream" author="examples">
        <pro-snowflake:createStream streamName="DYNAMIC_TABLE_STREAM"
                                    sourceName="SOURCE_DYNAMIC_TABLE">
            <pro-snowflake:dynamicTableStream/>
        </pro-snowflake:createStream>
    </changeSet>

    <!-- Event table stream -->
    <changeSet id="create-event-table-stream" author="examples">
        <pro-snowflake:createStream streamName="EVENT_TABLE_STREAM"
                                    sourceName="SOURCE_EVENT_TABLE">
            <pro-snowflake:eventTableStream/>
        </pro-snowflake:createStream>
    </changeSet>

</databaseChangeLog>

{
  "databaseChangeLog": [
    {
      "changeSet": {
        "id": "create-table-stream",
        "author": "examples",
        "changes": [
          {
            "createStream": {
              "streamName": "TABLE_STREAM",
              "sourceName": "SOURCE_TABLE",
              "replaceIfExists": true,
              "copyGrants": true,
              "comment": "Stream for tracking table changes",
              "tableStream": {
                "showInitialRows": true,
                "appendOnly": true,
                "timeTravelClause": {
                  "timeTravelType": "AT",
                  "offset": "-1"
                }
              },
              "tags": [
                {
                  "entry": {
                    "key": "MYDB.PUBLIC.environment",
                    "value": "production"
                  }
                }
              ]
            }
          }
        ]
      }
    },
    {
      "changeSet": {
        "id": "create-view-stream",
        "author": "examples",
        "changes": [
          {
            "createStream": {
              "streamName": "VIEW_STREAM",
              "sourceName": "SOURCE_VIEW",
              "ifNotExists": true,
              "viewStream": {
                "appendOnly": true
              }
            }
          }
        ]
      }
    },
    {
      "changeSet": {
        "id": "create-stage-stream",
        "author": "examples",
        "changes": [
          {
            "createStream": {
              "streamName": "STAGE_STREAM",
              "sourceName": "DIRECTORY_STAGE",
              "comment": "Stream for directory table",
              "stageStream": {}
            }
          }
        ]
      }
    },
    {
      "changeSet": {
        "id": "create-external-table-stream",
        "author": "examples",
        "changes": [
          {
            "createStream": {
              "streamName": "EXTERNAL_TABLE_STREAM",
              "sourceName": "SOURCE_EXTERNAL_TABLE",
              "comment": "External table stream",
              "externalTableStream": {
                "insertOnly": true
              },
              "tags": [
                {
                  "entry": {
                    "key": "MYDB.PUBLIC.environment",
                    "value": "prod"
                  }
                }
              ]
            }
          }
        ]
      }
    },
    {
      "changeSet": {
        "id": "create-dynamic-table-stream",
        "author": "examples",
        "changes": [
          {
            "createStream": {
              "streamName": "DYNAMIC_TABLE_STREAM",
              "sourceName": "SOURCE_DYNAMIC_TABLE",
              "dynamicTableStream": {}
            }
          }
        ]
      }
    },
    {
      "changeSet": {
        "id": "create-event-table-stream",
        "author": "examples",
        "changes": [
          {
            "createStream": {
              "streamName": "EVENT_TABLE_STREAM",
              "sourceName": "SOURCE_EVENT_TABLE",
              "eventTableStream": {}
            }
          }
        ]
      }
    }
  ]
}

databaseChangeLog:
  #Table stream with time travel
  - changeSet:
      id: create-table-stream
      author: examples
      changes:
        - createStream:
            streamName: TABLE_STREAM
            sourceName: SOURCE_TABLE
            replaceIfExists: true
            copyGrants: true
            comment: Stream for tracking table changes
            tableStream:
              showInitialRows: true
              appendOnly: true
              timeTravelClause:
                timeTravelType: AT
                offset: "-1"
            tags:
              - entry:
                  key: MYDB.PUBLIC.environment
                  value: production

  #View stream
  - changeSet:
      id: create-view-stream
      author: examples
      changes:
        - createStream:
            streamName: VIEW_STREAM
            sourceName: SOURCE_VIEW
            ifNotExists: true
            viewStream:
              appendOnly: true

  #Stage stream
  - changeSet:
      id: create-stage-stream
      author: examples
      changes:
        - createStream:
            streamName: STAGE_STREAM
            sourceName: DIRECTORY_STAGE
            comment: Stream for directory table
            stageStream: {}

  #External table stream with tags
  - changeSet:
      id: create-external-table-stream
      author: examples
      changes:
        - createStream:
            streamName: EXTERNAL_TABLE_STREAM
            sourceName: SOURCE_EXTERNAL_TABLE
            comment: External table stream
            externalTableStream:
              insertOnly: true
            tags:
              - entry:
                  key: MYDB.PUBLIC.environment
                  value: prod

  #Dynamic table stream
  - changeSet:
      id: create-dynamic-table-stream
      author: examples
      changes:
        - createStream:
            streamName: DYNAMIC_TABLE_STREAM
            sourceName: SOURCE_DYNAMIC_TABLE
            dynamicTableStream: {}

  #Event table stream
  - changeSet:
      id: create-event-table-stream
      author: examples
      changes:
        - createStream:
            streamName: EVENT_TABLE_STREAM
            sourceName: SOURCE_EVENT_TABLE
            eventTableStream: {}

--liquibase formatted sql

--changeset examples:table-stream-with-time-travel-and-tags
-- Table stream with time travel and tags
CREATE OR REPLACE STREAM MYDB.PUBLIC.TABLE_STREAM
WITH TAG (MYDB.PUBLIC.environment = 'production')
COPY GRANTS
ON TABLE MYDB.PUBLIC.SOURCE_TABLE
AT (OFFSET => -1)
APPEND_ONLY = TRUE
SHOW_INITIAL_ROWS = TRUE
COMMENT = 'Stream for tracking table changes';

--changeset examples:view-stream
-- View stream
CREATE STREAM IF NOT EXISTS MYDB.PUBLIC.VIEW_STREAM
ON VIEW MYDB.PUBLIC.SOURCE_VIEW
APPEND_ONLY = TRUE;

--changeset examples:stage-stream
-- Stage stream
CREATE STREAM MYDB.PUBLIC.STAGE_STREAM
ON STAGE MYDB.PUBLIC.DIRECTORY_STAGE
COMMENT = 'Stream for directory table';

--changeset examples:external-table-stream-with-tags
-- External table stream with tags
CREATE STREAM MYDB.PUBLIC.EXTERNAL_TABLE_STREAM
WITH TAG (MYDB.PUBLIC.environment = 'prod')
ON EXTERNAL TABLE MYDB.PUBLIC.SOURCE_EXTERNAL_TABLE
INSERT_ONLY = TRUE
COMMENT = 'External table stream';

--changeset examples:dynamic-table-stream
-- Dynamic table stream
CREATE STREAM MYDB.PUBLIC.DYNAMIC_TABLE_STREAM
ON DYNAMIC TABLE MYDB.PUBLIC.SOURCE_DYNAMIC_TABLE;

--changeset examples:event-table-stream
-- Event table stream
CREATE STREAM MYDB.PUBLIC.EVENT_TABLE_STREAM
ON EVENT TABLE MYDB.PUBLIC.SOURCE_EVENT_TABLE;

createStream

Available attributes

Stream type elements

tableStream / viewStream nested attributes

externalTableStream nested attributes

timeTravelClause nested attributes

XML

JSON

YAML

SQL

More like this