Repository: chanzuckerberg/terraform-provider-snowflake
Pull request #575 (2021-06-15 23:40:38)

feat: Add SHOW_INITIAL_ROWS to stream resource
<!-- Feel free to delete comments as you fill this in -->

<!-- summary of changes -->
Addressing this feature request https://github.com/chanzuckerberg/terraform-provider-snowflake/issues/537
## Test Plan
<!-- detail ways in which this PR has been tested or needs to be tested -->
* [ ] acceptance tests
<!-- add more below if you think they are relevant -->
* [ ] …

## References
<!-- issues documentation links, etc  -->

* 

github-actions[bot]: Integration tests failure for [bf28f637aadfd0dfdb80d58957b2c486bb620240](https://github.com/chanzuckerberg/terraform-provider-snowflake/actions/runs/941063577)
github-actions[bot]: Integration tests failure for [f65224dcad6d1111cd814d7dcf51174550a8333e](https://github.com/chanzuckerberg/terraform-provider-snowflake/actions/runs/941089544)
github-actions[bot]: Integration tests failure for [5626435065ec2dd36144a5b5230e7e1d7adbc18b](https://github.com/chanzuckerberg/terraform-provider-snowflake/actions/runs/941095658)
github-actions[bot]: Integration tests failure for [d85d147731c4784e247580fd075ebed2e096f493](https://github.com/chanzuckerberg/terraform-provider-snowflake/actions/runs/941102490)
github-actions[bot]: Integration tests success for [3497e6431fc48caf4a584063277f722b4fea5293](https://github.com/chanzuckerberg/terraform-provider-snowflake/actions/runs/941108680)
github-actions[bot]: Integration tests success for [909aee3f281910fc6b62ecf788080c230a36c79c](https://github.com/chanzuckerberg/terraform-provider-snowflake/actions/runs/941116466)
github-actions[bot]: Integration tests failure for [1eebe69841d7699d12503b387df5c9c1c519fb03](https://github.com/chanzuckerberg/terraform-provider-snowflake/actions/runs/943795388)
github-actions[bot]: Integration tests failure for [1bd48953ceeadd674600e3aff95537ba3b57e334](https://github.com/chanzuckerberg/terraform-provider-snowflake/actions/runs/943797446)
github-actions[bot]: Integration tests failure for [d853bf5624332780adfe40493d76f72aff8eb9a4](https://github.com/chanzuckerberg/terraform-provider-snowflake/actions/runs/943820164)
github-actions[bot]: Integration tests failure for [1d0e07f8018d89aca4fbba59fac4cc5a0a21b61c](https://github.com/chanzuckerberg/terraform-provider-snowflake/actions/runs/943834366)
github-actions[bot]: Integration tests failure for [06121354f5963857626153850c1d94683f17b108](https://github.com/chanzuckerberg/terraform-provider-snowflake/actions/runs/944247785)
github-actions[bot]: Integration tests failure for [967b3e2c7b18e3a507309f98e3ac6d67c5fa2df7](https://github.com/chanzuckerberg/terraform-provider-snowflake/actions/runs/944287346)
github-actions[bot]: Integration tests failure for [992b871095871e07c6e6b9b9b8ea3592b7e036b0](https://github.com/chanzuckerberg/terraform-provider-snowflake/actions/runs/944366363)
github-actions[bot]: Integration tests failure for [8551cf28a61fe091f2df22ff0d4ec3ddfb7cafad](https://github.com/chanzuckerberg/terraform-provider-snowflake/actions/runs/944391516)
github-actions[bot]: Integration tests failure for [53a9c4a06335d2afbbd08d04d7aff892da14fdc4](https://github.com/chanzuckerberg/terraform-provider-snowflake/actions/runs/944412897)
github-actions[bot]: Integration tests failure for [d8d3cb1657791cbdc001c6698e6ece7a9a0be494](https://github.com/chanzuckerberg/terraform-provider-snowflake/actions/runs/944566833)
github-actions[bot]: Integration tests failure for [570d9790b1569e2dfcd1191a1540c794ea73ab8c](https://github.com/chanzuckerberg/terraform-provider-snowflake/actions/runs/944592169)
github-actions[bot]: Integration tests success for [c20832cc985095a3ddbed38b80baf4b329750c2e](https://github.com/chanzuckerberg/terraform-provider-snowflake/actions/runs/944625864)
github-actions[bot]: Integration tests success for [40b0c5550f2afacbbe55754eb8834e8e7736d22c](https://github.com/chanzuckerberg/terraform-provider-snowflake/actions/runs/944748171)
github-actions[bot]: Integration tests success for [e23cf7c266f5537b5e52d4b208f87dc0f549ad72](https://github.com/chanzuckerberg/terraform-provider-snowflake/actions/runs/947240044)
github-actions[bot]: Integration tests success for [2c3b0723837d9a0fb2f7b5254799b173d72e2d54](https://github.com/chanzuckerberg/terraform-provider-snowflake/actions/runs/947243389)
github-actions[bot]: Integration tests success for [96e09248bbaeae94bc999ad83d32dd170b24c41e](https://github.com/chanzuckerberg/terraform-provider-snowflake/actions/runs/947248850)

Patch:
diff --git a/docs/resources/stream.md b/docs/resources/stream.md
index 6590c59b..e303aa78 100644
--- a/docs/resources/stream.md
+++ b/docs/resources/stream.md
@@ -42,6 +42,7 @@ resource snowflake_stream stream {
 - **comment** (String) Specifies a comment for the stream.
 - **id** (String) The ID of this resource.
 - **on_table** (String) Name of the table the stream will monitor.
+- **show_initial_rows** (Boolean) Specifies whether to return all existing rows in the source table as row inserts the first time the stream is consumed.
 
 ### Read-Only
 
diff --git a/pkg/resources/stream.go b/pkg/resources/stream.go
index f6c480ff..437685e0 100644
--- a/pkg/resources/stream.go
+++ b/pkg/resources/stream.go
@@ -45,14 +45,23 @@ var streamSchema = map[string]*schema.Schema{
 	"on_table": {
 		Type:        schema.TypeString,
 		Optional:    true,
+		ForceNew:    true,
 		Description: "Name of the table the stream will monitor.",
 	},
 	"append_only": {
 		Type:        schema.TypeBool,
 		Optional:    true,
+		ForceNew:    true,
 		Default:     false,
 		Description: "Type of the stream that will be created.",
 	},
+	"show_initial_rows": {
+		Type:        schema.TypeBool,
+		Optional:    true,
+		ForceNew:    true,
+		Default:     false,
+		Description: "Specifies whether to return all existing rows in the source table as row inserts the first time the stream is consumed.",
+	},
 	"owner": {
 		Type:        schema.TypeString,
 		Computed:    true,
@@ -160,6 +169,7 @@ func CreateStream(d *schema.ResourceData, meta interface{}) error {
 	name := d.Get("name").(string)
 	onTable := d.Get("on_table").(string)
 	appendOnly := d.Get("append_only").(bool)
+	showInitialRows := d.Get("show_initial_rows").(bool)
 
 	builder := snowflake.Stream(name, database, schema)
 
@@ -170,6 +180,7 @@ func CreateStream(d *schema.ResourceData, meta interface{}) error {
 
 	builder.WithOnTable(resultOnTable.DatabaseName, resultOnTable.SchemaName, resultOnTable.OnTableName)
 	builder.WithAppendOnly(appendOnly)
+	builder.WithShowInitialRows(showInitialRows)
 
 	// Set optionals
 	if v, ok := d.GetOk("comment"); ok {
@@ -226,6 +237,26 @@ func ReadStream(d *schema.ResourceData, meta interface{}) error {
 		return err
 	}
 
+	err = d.Set("on_table", stream.TableName.String)
+	if err != nil {
+		return err
+	}
+
+	err = d.Set("append_only", stream.AppendOnly)
+	if err != nil {
+		return err
+	}
+
+	err = d.Set("show_initial_rows", stream.ShowInitialRows)
+	if err != nil {
+		return err
+	}
+
+	err = d.Set("comment", stream.Comment.String)
+	if err != nil {
+		return err
+	}
+
 	err = d.Set("owner", stream.Owner.String)
 	if err != nil {
 		return err
diff --git a/pkg/snowflake/stream.go b/pkg/snowflake/stream.go
index 536b651c..b4ae4206 100644
--- a/pkg/snowflake/stream.go
+++ b/pkg/snowflake/stream.go
@@ -10,12 +10,13 @@ import (
 
 // 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
+	name            string
+	db              string
+	schema          string
+	onTable         string
+	appendOnly      bool
+	showInitialRows bool
+	comment         string
 }
 
 // QualifiedName prepends the db and schema if set and escapes everything nicely
@@ -50,12 +51,12 @@ func (sb *StreamBuilder) WithOnTable(d string, s string, t string) *StreamBuilde
 }
 
 func (sb *StreamBuilder) WithAppendOnly(b bool) *StreamBuilder {
-	sb.appendOnly = false
-
-	if b {
-		sb.appendOnly = b
-	}
+	sb.appendOnly = b
+	return sb
+}
 
+func (sb *StreamBuilder) WithShowInitialRows(b bool) *StreamBuilder {
+	sb.showInitialRows = b
 	return sb
 }
 
@@ -89,6 +90,8 @@ func (sb *StreamBuilder) Create() string {
 
 	q.WriteString(fmt.Sprintf(` APPEND_ONLY = %v`, sb.appendOnly))
 
+	q.WriteString(fmt.Sprintf(` SHOW_INITIAL_ROWS = %v`, sb.showInitialRows))
+
 	return q.String()
 }
 
@@ -113,16 +116,18 @@ func (sb *StreamBuilder) Show() string {
 }
 
 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"`
+	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"`
+	AppendOnly      bool           `db:"append_only"`
+	ShowInitialRows bool           `db:"show_initial_rows"`
+	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) {


Test patch:
diff --git a/pkg/resources/stream_acceptance_test.go b/pkg/resources/stream_acceptance_test.go
index fa36d8f3..b2efcbbd 100644
--- a/pkg/resources/stream_acceptance_test.go
+++ b/pkg/resources/stream_acceptance_test.go
@@ -23,7 +23,8 @@ func TestAcc_Stream(t *testing.T) {
 					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),
+					checkBool("snowflake_stream.test_stream", "append_only", false),
+					checkBool("snowflake_stream.test_stream", "show_initial_rows", false),
 				),
 			},
 		},
@@ -64,8 +65,6 @@ resource "snowflake_stream" "test_stream" {
 	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_grant_acceptance_test.go b/pkg/resources/stream_grant_acceptance_test.go
index d209583e..77694d23 100644
--- a/pkg/resources/stream_grant_acceptance_test.go
+++ b/pkg/resources/stream_grant_acceptance_test.go
@@ -96,11 +96,10 @@ resource "snowflake_stream" "test" {
 	name     = "{{ .stream_name }}"
 	comment  = "Terraform acceptance test"
 	on_table = "${snowflake_database.test.name}.${snowflake_schema.test.name}.${snowflake_table.test.name}"
-	append_only = true
 }
 
 resource "snowflake_stream_grant" "test" {
-    database_name = snowflake_database.test.name	
+  database_name = snowflake_database.test.name
 	roles         = [snowflake_role.test.name]
 	schema_name   = snowflake_schema.test.name
 	stream_name = snowflake_stream.test.name
@@ -140,7 +139,7 @@ resource "snowflake_role" "test" {
 }
 
 resource "snowflake_stream_grant" "test" {
-    database_name = snowflake_database.test.name	
+  database_name = snowflake_database.test.name
 	roles         = [snowflake_role.test.name]
 	schema_name   = snowflake_schema.test.name
 	on_future = true
diff --git a/pkg/resources/stream_test.go b/pkg/resources/stream_test.go
index 3f166a9a..f0535810 100644
--- a/pkg/resources/stream_test.go
+++ b/pkg/resources/stream_test.go
@@ -22,17 +22,18 @@ 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,
+		"name":              "stream_name",
+		"database":          "database_name",
+		"schema":            "schema_name",
+		"comment":           "great comment",
+		"on_table":          "target_db.target_schema.target_table",
+		"append_only":       true,
+		"show_initial_rows": 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))
+		mock.ExpectExec(`CREATE STREAM "database_name"."schema_name"."stream_name" ON TABLE "target_db"."target_schema"."target_table" COMMENT = 'great comment' APPEND_ONLY = true SHOW_INITIAL_ROWS = true`).WillReturnResult(sqlmock.NewResult(1, 1))
 		expectStreamRead(mock)
 		err := resources.CreateStream(d, db)
 		r.NoError(err)
@@ -48,14 +49,14 @@ func expectStreamRead(mock sqlmock.Sqlmock) {
 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"})
+	d := stream(t, "database_name|schema_name|stream_name", map[string]interface{}{"name": "stream_name", "comment": "grand 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))
+		r.Equal("grand comment", d.Get("comment").(string))
 
 		// Test when resource is not found, checking if state will be empty
 		r.NotEmpty(d.State())
diff --git a/pkg/snowflake/stream_test.go b/pkg/snowflake/stream_test.go
index 9b93f776..9ebadf66 100644
--- a/pkg/snowflake/stream_test.go
+++ b/pkg/snowflake/stream_test.go
@@ -11,13 +11,16 @@ func TestStreamCreate(t *testing.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`)
+	r.Equal(s.Create(), `CREATE STREAM "test_db"."test_schema"."test_stream" ON TABLE "test_db"."test_schema"."test_target_table" APPEND_ONLY = false SHOW_INITIAL_ROWS = 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`)
+	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 SHOW_INITIAL_ROWS = false`)
+
+	s.WithShowInitialRows(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 = false SHOW_INITIAL_ROWS = true`)
 
 	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`)
+	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 SHOW_INITIAL_ROWS = true`)
 }
 
 func TestStreamChangeComment(t *testing.T) {