Repository: chanzuckerberg/terraform-provider-snowflake
Pull request #286 (2020-10-26 02:34:48)

[feature] Snowflake Resource for Streams
**NOTE:**  Was approved from **Master** PR 280. Re-created it now from **Main** branch

## Summary of changes

1. Created a resource to support the creation of streams in Snowflake
2. Note: Resource for Stream Grants is not part of this PR.

**Sample Syntax:**
```hcl
resource "snowflake_stream" "test_stream" {
	database = "your_database_name"
	schema   = "your_schema_name"
	name     = "your_new_stream"
	comment  = "Terraform acceptance test"
	on_table = "target_db.target_schema.target_table"
	append_only = true
	
}
```


## Test Plan

1. Internal and Unit test was implemented and done.
2. Acceptance test was done using our own Snowflake account. **make test-acceptance** was executed from the terminal
3. Sanity check was done by creating a local build of the provider and installing the plugin in our terraform project to test the creation of the streams

## References
- https://docs.snowflake.com/en/sql-reference/sql/create-stream.html

## Linked Issue
- #273 


Patch:
diff --git a/docs/resources/stream.md b/docs/resources/stream.md
new file mode 100644
index 00000000..6f595d9d
--- /dev/null
+++ b/docs/resources/stream.md
@@ -0,0 +1,16 @@
+
+# snowflake_stream
+
+<!-- These docs are auto-generated by code in ./docgen, run by with make docs. Manual edits will be overwritten. -->
+
+## properties
+
+|    NAME     |  TYPE  |                                                     DESCRIPTION                                                     | OPTIONAL | REQUIRED  | COMPUTED | DEFAULT |
+|-------------|--------|---------------------------------------------------------------------------------------------------------------------|----------|-----------|----------|---------|
+| append_only | bool   | Type of the stream that will be created.                                                                            | true     | false     | false    | false   |
+| comment     | string | Specifies a comment for the stream.                                                                                 | true     | false     | false    |         |
+| database    | string | The database in which to create the stream.                                                                         | false    | true      | false    |         |
+| name        | string | Specifies the identifier for the stream; must be unique for the database and schema in which the stream is created. | false    | true      | false    |         |
+| on_table    | string | Name of the table the stream will monitor.                                                                          | true     | false     | false    |         |
+| owner       | string | Name of the role that owns the stream.                                                                              | false    | false     | true     |         |
+| schema      | string | The schema in which to create the stream.                                                                           | false    | true      | false    |         |
diff --git a/pkg/provider/provider.go b/pkg/provider/provider.go
index ddb45693..f168f9ea 100644
--- a/pkg/provider/provider.go
+++ b/pkg/provider/provider.go
@@ -86,6 +86,7 @@ func Provider() *schema.Provider {
 			"snowflake_stage":                     resources.Stage(),
 			"snowflake_stage_grant":               resources.StageGrant(),
 			"snowflake_storage_integration":       resources.StorageIntegration(),
+			"snowflake_stream":                    resources.Stream(),
 			"snowflake_user":                      resources.User(),
 			"snowflake_view":                      resources.View(),
 			"snowflake_view_grant":                resources.ViewGrant(),
diff --git a/pkg/resources/stream.go b/pkg/resources/stream.go
new file mode 100644
index 00000000..39f6bc93
--- /dev/null
+++ b/pkg/resources/stream.go
@@ -0,0 +1,284 @@
+package resources
+
+import (
+	"bytes"
+	"database/sql"
+
+	"github.com/chanzuckerberg/terraform-provider-snowflake/pkg/snowflake"
+	"github.com/hashicorp/terraform-plugin-sdk/helper/schema"
+	"github.com/pkg/errors"
+
+	"encoding/csv"
+	"fmt"
+	"strings"
+)
+
+const (
+	streamIDDelimiter        = '|'
+	streamOnTableIDDelimiter = '.'
+)
+
+var streamSchema = map[string]*schema.Schema{
+	"name": {
+		Type:        schema.TypeString,
+		Required:    true,
+		ForceNew:    true,
+		Description: "Specifies the identifier for the stream; must be unique for the database and schema in which the stream is created.",
+	},
+	"schema": {
+		Type:        schema.TypeString,
+		Required:    true,
+		ForceNew:    true,
+		Description: "The schema in which to create the stream.",
+	},
+	"database": {
+		Type:        schema.TypeString,
+		Required:    true,
+		ForceNew:    true,
+		Description: "The database in which to create the stream.",
+	},
+	"comment": {
+		Type:        schema.TypeString,
+		Optional:    true,
+		Description: "Specifies a comment for the stream.",
+	},
+	"on_table": {
+		Type:        schema.TypeString,
+		Optional:    true,
+		Description: "Name of the table the stream will monitor.",
+	},
+	"append_only": {
+		Type:        schema.TypeBool,
+		Optional:    true,
+		Default:     false,
+		Description: "Type of the stream that will be created.",
+	},
+	"owner": {
+		Type:        schema.TypeString,
+		Computed:    true,
+		Description: "Name of the role that owns the stream.",
+	},
+}
+
+func Stream() *schema.Resource {
+	return &schema.Resource{
+		Create: CreateStream,
+		Read:   ReadStream,
+		Update: UpdateStream,
+		Delete: DeleteStream,
+
+		Schema: streamSchema,
+		Importer: &schema.ResourceImporter{
+			State: schema.ImportStatePassthrough,
+		},
+	}
+}
+
+type streamID struct {
+	DatabaseName string
+	SchemaName   string
+	StreamName   string
+}
+
+type streamOnTableID struct {
+	DatabaseName string
+	SchemaName   string
+	OnTableName  string
+}
+
+//String() takes in a streamID object and returns a pipe-delimited string:
+//DatabaseName|SchemaName|StreamName
+func (si *streamID) String() (string, error) {
+	var buf bytes.Buffer
+	csvWriter := csv.NewWriter(&buf)
+	csvWriter.Comma = streamIDDelimiter
+	dataIdentifiers := [][]string{{si.DatabaseName, si.SchemaName, si.StreamName}}
+	err := csvWriter.WriteAll(dataIdentifiers)
+	if err != nil {
+		return "", err
+	}
+	strStreamID := strings.TrimSpace(buf.String())
+	return strStreamID, nil
+}
+
+// streamIDFromString() takes in a pipe-delimited string: DatabaseName|SchemaName|StreamName
+// and returns a streamID object
+func streamIDFromString(stringID string) (*streamID, error) {
+	reader := csv.NewReader(strings.NewReader(stringID))
+	reader.Comma = streamIDDelimiter
+	lines, err := reader.ReadAll()
+	if err != nil {
+		return nil, fmt.Errorf("Not CSV compatible")
+	}
+
+	if len(lines) != 1 {
+		return nil, fmt.Errorf("1 line at a time")
+	}
+	if len(lines[0]) != 3 {
+		return nil, fmt.Errorf("3 fields allowed")
+	}
+
+	streamResult := &streamID{
+		DatabaseName: lines[0][0],
+		SchemaName:   lines[0][1],
+		StreamName:   lines[0][2],
+	}
+	return streamResult, nil
+}
+
+// streamOnTableIDFromString() takes in a dot-delimited string: DatabaseName.SchemaName.TableName
+// and returns a streamOnTableID object
+func streamOnTableIDFromString(stringID string) (*streamOnTableID, error) {
+	reader := csv.NewReader(strings.NewReader(stringID))
+	reader.Comma = streamOnTableIDDelimiter
+	lines, err := reader.ReadAll()
+	if err != nil {
+		return nil, fmt.Errorf("Not CSV compatible")
+	}
+
+	if len(lines) != 1 {
+		return nil, fmt.Errorf("1 line at a time")
+	}
+	if len(lines[0]) != 3 {
+		//return nil, fmt.Errorf("on table format: database_name.schema_name.target_table_name")
+		return nil, fmt.Errorf("invalid format for on_table: %v , expected: <database_name.schema_name.target_table_name>", strings.Join(lines[0], "."))
+	}
+
+	streamOnTableResult := &streamOnTableID{
+		DatabaseName: lines[0][0],
+		SchemaName:   lines[0][1],
+		OnTableName:  lines[0][2],
+	}
+	return streamOnTableResult, nil
+}
+
+// CreateStream implements schema.CreateFunc
+func CreateStream(data *schema.ResourceData, meta interface{}) error {
+	db := meta.(*sql.DB)
+	database := data.Get("database").(string)
+	schema := data.Get("schema").(string)
+	name := data.Get("name").(string)
+	onTable := data.Get("on_table").(string)
+	appendOnly := data.Get("append_only").(bool)
+
+	builder := snowflake.Stream(name, database, schema)
+
+	resultOnTable, err := streamOnTableIDFromString(onTable)
+	if err != nil {
+		return err
+	}
+
+	builder.WithOnTable(resultOnTable.DatabaseName, resultOnTable.SchemaName, resultOnTable.OnTableName)
+	builder.WithAppendOnly(appendOnly)
+
+	// Set optionals
+	if v, ok := data.GetOk("comment"); ok {
+		builder.WithComment(v.(string))
+	}
+
+	stmt := builder.Create()
+	err = snowflake.Exec(db, stmt)
+	if err != nil {
+		return errors.Wrapf(err, "error creating stream %v", name)
+	}
+
+	streamID := &streamID{
+		DatabaseName: database,
+		SchemaName:   schema,
+		StreamName:   name,
+	}
+	dataIDInput, err := streamID.String()
+	if err != nil {
+		return err
+	}
+	data.SetId(dataIDInput)
+
+	return ReadStream(data, meta)
+}
+
+// ReadStream implements schema.ReadFunc
+func ReadStream(data *schema.ResourceData, meta interface{}) error {
+	db := meta.(*sql.DB)
+	streamID, err := streamIDFromString(data.Id())
+	if err != nil {
+		return err
+	}
+
+	dbName := streamID.DatabaseName
+	schema := streamID.SchemaName
+	name := streamID.StreamName
+
+	stmt := snowflake.Stream(name, dbName, schema).Show()
+	row := snowflake.QueryRow(db, stmt)
+	stream, err := snowflake.ScanStream(row)
+	if err != nil {
+		return err
+	}
+
+	err = data.Set("name", stream.StreamName.String)
+	if err != nil {
+		return err
+	}
+
+	err = data.Set("owner", stream.Owner.String)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// DeleteStream implements schema.DeleteFunc
+func DeleteStream(data *schema.ResourceData, meta interface{}) error {
+	db := meta.(*sql.DB)
+	streamID, err := streamIDFromString(data.Id())
+	if err != nil {
+		return err
+	}
+
+	dbName := streamID.DatabaseName
+	schema := streamID.SchemaName
+	streamName := streamID.StreamName
+
+	q := snowflake.Stream(streamName, dbName, schema).Drop()
+
+	err = snowflake.Exec(db, q)
+	if err != nil {
+		return errors.Wrapf(err, "error deleting stream %v", data.Id())
+	}
+
+	data.SetId("")
+
+	return nil
+}
+
+// UpdateStream implements schema.UpdateFunc
+func UpdateStream(data *schema.ResourceData, meta interface{}) error {
+	// https://www.terraform.io/docs/extend/writing-custom-providers.html#error-handling-amp-partial-state
+	data.Partial(true)
+
+	streamID, err := streamIDFromString(data.Id())
+	if err != nil {
+		return err
+	}
+
+	dbName := streamID.DatabaseName
+	schema := streamID.SchemaName
+	streamName := streamID.StreamName
+
+	builder := snowflake.Stream(streamName, dbName, schema)
+
+	db := meta.(*sql.DB)
+	if data.HasChange("comment") {
+		_, comment := data.GetChange("comment")
+		q := builder.ChangeComment(comment.(string))
+		err := snowflake.Exec(db, q)
+		if err != nil {
+			return errors.Wrapf(err, "error updating stream comment on %v", data.Id())
+		}
+
+		data.SetPartial("comment")
+	}
+
+	return ReadStream(data, meta)
+}
diff --git a/pkg/snowflake/stream.go b/pkg/snowflake/stream.go
new file mode 100644
index 00000000..536b651c
--- /dev/null
+++ b/pkg/snowflake/stream.go
@@ -0,0 +1,132 @@
+package snowflake
+
+import (
+	"database/sql"
+	"fmt"
+	"strings"
+
+	"github.com/jmoiron/sqlx"
+)
+
+// StreamBuilder abstracts the creation of SQL queries for a Snowflake stream
+type StreamBuilder struct {
+	name       string
+	db         string
+	schema     string
+	onTable    string
+	appendOnly bool
+	comment    string
+}
+
+// QualifiedName prepends the db and schema if set and escapes everything nicely
+func (sb *StreamBuilder) QualifiedName() string {
+	var n strings.Builder
+
+	if sb.db != "" && sb.schema != "" {
+		n.WriteString(fmt.Sprintf(`"%v"."%v".`, sb.db, sb.schema))
+	}
+
+	if sb.db != "" && sb.schema == "" {
+		n.WriteString(fmt.Sprintf(`"%v"..`, sb.db))
+	}
+
+	if sb.db == "" && sb.schema != "" {
+		n.WriteString(fmt.Sprintf(`"%v".`, sb.schema))
+	}
+
+	n.WriteString(fmt.Sprintf(`"%v"`, sb.name))
+
+	return n.String()
+}
+
+func (sb *StreamBuilder) WithComment(c string) *StreamBuilder {
+	sb.comment = c
+	return sb
+}
+
+func (sb *StreamBuilder) WithOnTable(d string, s string, t string) *StreamBuilder {
+	sb.onTable = fmt.Sprintf(`"%v"."%v"."%v"`, d, s, t)
+	return sb
+}
+
+func (sb *StreamBuilder) WithAppendOnly(b bool) *StreamBuilder {
+	sb.appendOnly = false
+
+	if b {
+		sb.appendOnly = b
+	}
+
+	return sb
+}
+
+// Stream returns a pointer to a Builder that abstracts the DDL operations for a stream.
+//
+// Supported DDL operations are:
+//   - CREATE Stream
+//   - ALTER Stream
+//	 - DROP Stream
+//   - SHOW Stream
+//
+// [Snowflake Reference](https://docs.snowflake.com/en/sql-reference/sql/create-stream.html)
+func Stream(name, db, schema string) *StreamBuilder {
+	return &StreamBuilder{
+		name:   name,
+		db:     db,
+		schema: schema,
+	}
+}
+
+// Create returns the SQL statement required to create a stream
+func (sb *StreamBuilder) Create() string {
+	q := strings.Builder{}
+	q.WriteString(fmt.Sprintf(`CREATE STREAM %v`, sb.QualifiedName()))
+
+	q.WriteString(fmt.Sprintf(` ON TABLE %v`, sb.onTable))
+
+	if sb.comment != "" {
+		q.WriteString(fmt.Sprintf(` COMMENT = '%v'`, EscapeString(sb.comment)))
+	}
+
+	q.WriteString(fmt.Sprintf(` APPEND_ONLY = %v`, sb.appendOnly))
+
+	return q.String()
+}
+
+// ChangeComment returns the SQL query that will update the comment on the stream.
+func (sb *StreamBuilder) ChangeComment(c string) string {
+	return fmt.Sprintf(`ALTER STREAM %v SET COMMENT = '%v'`, sb.QualifiedName(), EscapeString(c))
+}
+
+// RemoveComment returns the SQL query that will remove the comment on the stream.
+func (sb *StreamBuilder) RemoveComment() string {
+	return fmt.Sprintf(`ALTER STREAM %v UNSET COMMENT`, sb.QualifiedName())
+}
+
+// Drop returns the SQL query that will drop a stream.
+func (sb *StreamBuilder) Drop() string {
+	return fmt.Sprintf(`DROP STREAM %v`, sb.QualifiedName())
+}
+
+// Show returns the SQL query that will show a stream.
+func (sb *StreamBuilder) Show() string {
+	return fmt.Sprintf(`SHOW STREAMS LIKE '%v' IN DATABASE "%v"`, sb.name, sb.db)
+}
+
+type descStreamRow struct {
+	CreatedOn    sql.NullString `db:"created_on"`
+	StreamName   sql.NullString `db:"name"`
+	DatabaseName sql.NullString `db:"database_name"`
+	SchemaName   sql.NullString `db:"schema_name"`
+	Owner        sql.NullString `db:"owner"`
+	Comment      sql.NullString `db:"comment"`
+	TableName    sql.NullString `db:"table_name"`
+	Type         sql.NullString `db:"type"`
+	Stale        sql.NullString `db:"stale"`
+	Mode         sql.NullString `db:"mode"`
+}
+
+func ScanStream(row *sqlx.Row) (*descStreamRow, error) {
+	t := &descStreamRow{}
+	e := row.StructScan(t)
+	return t, e
+}


Test patch:
diff --git a/pkg/resources/helpers_test.go b/pkg/resources/helpers_test.go
index 21691830..99f4e7bb 100644
--- a/pkg/resources/helpers_test.go
+++ b/pkg/resources/helpers_test.go
@@ -82,6 +82,14 @@ func accountGrant(t *testing.T, id string, params map[string]interface{}) *schem
 	return d
 }
 
+func stream(t *testing.T, id string, params map[string]interface{}) *schema.ResourceData {
+	r := require.New(t)
+	d := schema.TestResourceDataRaw(t, resources.Stream().Schema, params)
+	r.NotNil(d)
+	d.SetId(id)
+	return d
+}
+
 func providers() map[string]terraform.ResourceProvider {
 	p := provider.Provider()
 	return map[string]terraform.ResourceProvider{
diff --git a/pkg/resources/stream_acceptance_test.go b/pkg/resources/stream_acceptance_test.go
new file mode 100644
index 00000000..c9b05c1b
--- /dev/null
+++ b/pkg/resources/stream_acceptance_test.go
@@ -0,0 +1,71 @@
+package resources_test
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/hashicorp/terraform-plugin-sdk/helper/acctest"
+	"github.com/hashicorp/terraform-plugin-sdk/helper/resource"
+)
+
+func TestAccStream(t *testing.T) {
+	accName := acctest.RandStringFromCharSet(10, acctest.CharSetAlpha)
+
+	resource.Test(t, resource.TestCase{
+		Providers: providers(),
+		Steps: []resource.TestStep{
+			{
+				Config: streamConfig(accName),
+				Check: resource.ComposeTestCheckFunc(
+					resource.TestCheckResourceAttr("snowflake_stream.test_stream", "name", accName),
+					resource.TestCheckResourceAttr("snowflake_stream.test_stream", "database", accName),
+					resource.TestCheckResourceAttr("snowflake_stream.test_stream", "schema", accName),
+					resource.TestCheckResourceAttr("snowflake_stream.test_stream", "on_table", fmt.Sprintf("%s.%s.%s", accName, accName, "stream_on_table")),
+					resource.TestCheckResourceAttr("snowflake_stream.test_stream", "comment", "Terraform acceptance test"),
+					checkBool("snowflake_stream.test_stream", "append_only", true),
+				),
+			},
+		},
+	})
+}
+
+func streamConfig(name string) string {
+	s := `
+resource "snowflake_database" "test_database" {
+	name    = "%s"
+	comment = "Terraform acceptance test"
+}
+
+resource "snowflake_schema" "test_schema" {
+	name     = "%s"
+	database = snowflake_database.test_database.name
+	comment  = "Terraform acceptance test"
+}
+
+resource "snowflake_table" "test_stream_on_table" {
+	database = snowflake_database.test_database.name
+	schema   = snowflake_schema.test_schema.name
+	name     = "stream_on_table"
+	comment  = "Terraform acceptance test"
+	column {
+		name = "column1"
+		type = "VARIANT"
+	}
+	column {
+		name = "column2"
+		type = "VARCHAR"
+	}
+}
+
+resource "snowflake_stream" "test_stream" {
+	database = snowflake_database.test_database.name
+	schema   = snowflake_schema.test_schema.name
+	name     = "%s"
+	comment  = "Terraform acceptance test"
+	on_table = "${snowflake_database.test_database.name}.${snowflake_schema.test_schema.name}.${snowflake_table.test_stream_on_table.name}"
+	append_only = true
+	
+}
+`
+	return fmt.Sprintf(s, name, name, name)
+}
diff --git a/pkg/resources/stream_internal_test.go b/pkg/resources/stream_internal_test.go
new file mode 100644
index 00000000..5d76a56d
--- /dev/null
+++ b/pkg/resources/stream_internal_test.go
@@ -0,0 +1,104 @@
+package resources
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestStreamIDFromString(t *testing.T) {
+	r := require.New(t)
+	// Vanilla
+	id := "database_name|schema_name|stream"
+	stream, err := streamIDFromString(id)
+	r.NoError(err)
+	r.Equal("database_name", stream.DatabaseName)
+	r.Equal("schema_name", stream.SchemaName)
+	r.Equal("stream", stream.StreamName)
+
+	// Bad ID -- not enough fields
+	id = "database"
+	_, err = streamIDFromString(id)
+	r.Equal(fmt.Errorf("3 fields allowed"), err)
+
+	// Bad ID
+	id = "||"
+	_, err = streamIDFromString(id)
+	r.NoError(err)
+
+	// 0 lines
+	id = ""
+	_, err = streamIDFromString(id)
+	r.Equal(fmt.Errorf("1 line at a time"), err)
+
+	// 2 lines
+	id = `database_name|schema_name|stream
+	database_name|schema_name|stream`
+	_, err = streamIDFromString(id)
+	r.Equal(fmt.Errorf("1 line at a time"), err)
+}
+
+func TestStreamStruct(t *testing.T) {
+	r := require.New(t)
+
+	// Vanilla
+	stream := &streamID{
+		DatabaseName: "database_name",
+		SchemaName:   "schema_name",
+		StreamName:   "stream_name",
+	}
+	sID, err := stream.String()
+	r.NoError(err)
+	r.Equal("database_name|schema_name|stream_name", sID)
+
+	// Empty grant
+	stream = &streamID{}
+	sID, err = stream.String()
+	r.NoError(err)
+	r.Equal("||", sID)
+
+	// Grant with extra delimiters
+	stream = &streamID{
+		DatabaseName: "database|name",
+		StreamName:   "stream|name",
+	}
+	sID, err = stream.String()
+	r.NoError(err)
+	newStream, err := streamIDFromString(sID)
+	r.NoError(err)
+	r.Equal("database|name", newStream.DatabaseName)
+	r.Equal("stream|name", newStream.StreamName)
+}
+
+func TestStreamOnTableIDFromString(t *testing.T) {
+	r := require.New(t)
+	// Vanilla
+	id := "database_name.schema_name.target_table_name"
+	streamOnTable, err := streamOnTableIDFromString(id)
+	r.NoError(err)
+	r.Equal("database_name", streamOnTable.DatabaseName)
+	r.Equal("schema_name", streamOnTable.SchemaName)
+	r.Equal("target_table_name", streamOnTable.OnTableName)
+
+	// Bad ID -- not enough fields
+	id = "database.schema"
+	_, err = streamOnTableIDFromString(id)
+	r.Equal(fmt.Errorf("invalid format for on_table: database.schema , expected: <database_name.schema_name.target_table_name>"), err)
+
+	// Bad ID
+	id = ".."
+	_, err = streamOnTableIDFromString(id)
+	r.NoError(err)
+
+	// 0 lines
+	id = ""
+	_, err = streamOnTableIDFromString(id)
+	r.Equal(fmt.Errorf("1 line at a time"), err)
+
+	// 2 lines
+	id = `database_name.schema_name.target_table_name
+	database_name.schema_name.target_table_name`
+	_, err = streamOnTableIDFromString(id)
+	r.Equal(fmt.Errorf("1 line at a time"), err)
+}
diff --git a/pkg/resources/stream_test.go b/pkg/resources/stream_test.go
new file mode 100644
index 00000000..3b16eb8f
--- /dev/null
+++ b/pkg/resources/stream_test.go
@@ -0,0 +1,94 @@
+package resources_test
+
+import (
+	"database/sql"
+	"testing"
+
+	sqlmock "github.com/DATA-DOG/go-sqlmock"
+	"github.com/chanzuckerberg/terraform-provider-snowflake/pkg/provider"
+	"github.com/chanzuckerberg/terraform-provider-snowflake/pkg/resources"
+	. "github.com/chanzuckerberg/terraform-provider-snowflake/pkg/testhelpers"
+	"github.com/stretchr/testify/require"
+)
+
+func TestStream(t *testing.T) {
+	r := require.New(t)
+	err := resources.Stream().InternalValidate(provider.Provider().Schema, true)
+	r.NoError(err)
+}
+
+func TestStreamCreate(t *testing.T) {
+	r := require.New(t)
+
+	in := map[string]interface{}{
+		"name":        "stream_name",
+		"database":    "database_name",
+		"schema":      "schema_name",
+		"comment":     "great comment",
+		"on_table":    "target_db.target_schema.target_table",
+		"append_only": true,
+	}
+	d := stream(t, "database_name|schema_name|stream_name", in)
+
+	WithMockDb(t, func(db *sql.DB, mock sqlmock.Sqlmock) {
+		mock.ExpectExec(`CREATE STREAM "database_name"."schema_name"."stream_name" ON TABLE "target_db"."target_schema"."target_table" COMMENT = 'great comment' APPEND_ONLY = true`).WillReturnResult(sqlmock.NewResult(1, 1))
+		expectStreamRead(mock)
+		err := resources.CreateStream(d, db)
+		r.NoError(err)
+		r.Equal("stream_name", d.Get("name").(string))
+	})
+}
+
+func expectStreamRead(mock sqlmock.Sqlmock) {
+	rows := sqlmock.NewRows([]string{"name", "database_name", "schema_name", "owner", "comment", "table_name", "type", "stale", "mode"}).AddRow("stream_name", "database_name", "schema_name", "owner_name", "grand comment", "target_table", "DELTA", false, "APPEND_ONLY")
+	mock.ExpectQuery(`SHOW STREAMS LIKE 'stream_name' IN DATABASE "database_name"`).WillReturnRows(rows)
+}
+
+func TestStreamRead(t *testing.T) {
+	r := require.New(t)
+
+	d := stream(t, "database_name|schema_name|stream_name", map[string]interface{}{"name": "stream_name", "comment": "mock comment"})
+
+	WithMockDb(t, func(db *sql.DB, mock sqlmock.Sqlmock) {
+		expectStreamRead(mock)
+
+		err := resources.ReadStream(d, db)
+		r.NoError(err)
+		r.Equal("stream_name", d.Get("name").(string))
+		r.Equal("mock comment", d.Get("comment").(string))
+	})
+}
+
+func TestStreamDelete(t *testing.T) {
+	r := require.New(t)
+
+	d := stream(t, "database_name|schema_name|drop_it", map[string]interface{}{"name": "drop_it"})
+
+	WithMockDb(t, func(db *sql.DB, mock sqlmock.Sqlmock) {
+		mock.ExpectExec(`DROP STREAM "database_name"."schema_name"."drop_it"`).WillReturnResult(sqlmock.NewResult(1, 1))
+		err := resources.DeleteStream(d, db)
+		r.NoError(err)
+	})
+}
+
+func TestStreamUpdate(t *testing.T) {
+	r := require.New(t)
+
+	in := map[string]interface{}{
+		"name":        "stream_name",
+		"database":    "database_name",
+		"schema":      "schema_name",
+		"comment":     "new stream comment",
+		"on_table":    "target_table",
+		"append_only": true,
+	}
+
+	d := stream(t, "database_name|schema_name|stream_name", in)
+
+	WithMockDb(t, func(db *sql.DB, mock sqlmock.Sqlmock) {
+		mock.ExpectExec(`ALTER STREAM "database_name"."schema_name"."stream_name" SET COMMENT = 'new stream comment'`).WillReturnResult(sqlmock.NewResult(1, 1))
+		expectStreamRead(mock)
+		err := resources.UpdateStream(d, db)
+		r.NoError(err)
+	})
+}
diff --git a/pkg/snowflake/stream_test.go b/pkg/snowflake/stream_test.go
new file mode 100644
index 00000000..9b93f776
--- /dev/null
+++ b/pkg/snowflake/stream_test.go
@@ -0,0 +1,45 @@
+package snowflake
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestStreamCreate(t *testing.T) {
+	r := require.New(t)
+	s := Stream("test_stream", "test_db", "test_schema")
+
+	s.WithOnTable("test_db", "test_schema", "test_target_table")
+	r.Equal(s.Create(), `CREATE STREAM "test_db"."test_schema"."test_stream" ON TABLE "test_db"."test_schema"."test_target_table" APPEND_ONLY = false`)
+
+	s.WithComment("Test Comment")
+	r.Equal(s.Create(), `CREATE STREAM "test_db"."test_schema"."test_stream" ON TABLE "test_db"."test_schema"."test_target_table" COMMENT = 'Test Comment' APPEND_ONLY = false`)
+
+	s.WithAppendOnly(true)
+	r.Equal(s.Create(), `CREATE STREAM "test_db"."test_schema"."test_stream" ON TABLE "test_db"."test_schema"."test_target_table" COMMENT = 'Test Comment' APPEND_ONLY = true`)
+}
+
+func TestStreamChangeComment(t *testing.T) {
+	r := require.New(t)
+	s := Stream("test_stream", "test_db", "test_schema")
+	r.Equal(s.ChangeComment("new stream comment"), `ALTER STREAM "test_db"."test_schema"."test_stream" SET COMMENT = 'new stream comment'`)
+}
+
+func TestStreamRemoveComment(t *testing.T) {
+	r := require.New(t)
+	s := Stream("test_stream", "test_db", "test_schema")
+	r.Equal(s.RemoveComment(), `ALTER STREAM "test_db"."test_schema"."test_stream" UNSET COMMENT`)
+}
+
+func TestStreamDrop(t *testing.T) {
+	r := require.New(t)
+	s := Stream("test_stream", "test_db", "test_schema")
+	r.Equal(s.Drop(), `DROP STREAM "test_db"."test_schema"."test_stream"`)
+}
+
+func TestStreamShow(t *testing.T) {
+	r := require.New(t)
+	s := Stream("test_stream", "test_db", "test_schema")
+	r.Equal(s.Show(), `SHOW STREAMS LIKE 'test_stream' IN DATABASE "test_db"`)
+}