Repository: chanzuckerberg/terraform-provider-snowflake
Pull request #503 (2021-03-25 15:22:17)

Support adding, removing and changing the type of columns in tables
The current behaviour is to drop the table and all its data when a column is added, removed, or modified (#446, #421). This PR mostly fixes this. 

Now supported without having to drop and recreate the table:
* Adding columns
* Removing columns
* Changing the data type of a column

Partially supported:
* Renaming columns - the column is dropped and recreated leading to _some_ data loss. There is not currently any better way to correlate columns than their name.

## Test Plan
* [ ✔] acceptance tests (plus made the existing ones more rigorous)
* [✔ ] snowflake unit tests

## References
#446 
#421

[ALTER TABLE documentation](https://docs.snowflake.com/en/sql-reference/sql/alter-table.html)

AdamDewberry: Ooo this looks good, excited for a codeowner to review
alldoami: /ok-to-test sha=dc7fdab
alldoami: @edwardbrowncross could you update your branch with `main`?
github-actions[bot]: Integration tests success for [dc7fdab](https://github.com/chanzuckerberg/terraform-provider-snowflake/actions/runs/712384796)
alldoami: /ok-to-test sha=54fb130
github-actions[bot]: Integration tests success for [54fb130](https://github.com/chanzuckerberg/terraform-provider-snowflake/actions/runs/734096201)

Patch:
diff --git a/pkg/resources/table.go b/pkg/resources/table.go
index f8e56943..5db15aca 100644
--- a/pkg/resources/table.go
+++ b/pkg/resources/table.go
@@ -40,7 +40,6 @@ var tableSchema = map[string]*schema.Schema{
 		Type:        schema.TypeList,
 		Required:    true,
 		MinItems:    1,
-		ForceNew:    true,
 		Description: "Definitions of a column to create in the table. Minimum one required.",
 		Elem: &schema.Resource{
 			Schema: map[string]*schema.Schema{
@@ -129,6 +128,76 @@ func tableIDFromString(stringID string) (*tableID, error) {
 	return tableResult, nil
 }
 
+type column struct {
+	name     string
+	dataType string
+}
+
+func (c column) toSnowflakeColumn() snowflake.Column {
+	sC := snowflake.Column{}
+	return *sC.WithName(c.name).WithType(c.dataType)
+}
+
+type columns []column
+
+func (c columns) toSnowflakeColumns() []snowflake.Column {
+	sC := make([]snowflake.Column, len(c))
+	for i, col := range c {
+		sC[i] = col.toSnowflakeColumn()
+	}
+	return sC
+}
+
+func (old columns) getNewIn(new columns) (added columns) {
+	added = columns{}
+	for _, cO := range old {
+		found := false
+		for _, cN := range new {
+			if cO.name == cN.name {
+				found = true
+				break
+			}
+		}
+		if !found {
+			added = append(added, cO)
+		}
+	}
+	return
+}
+
+func (old columns) getChangedTypes(new columns) (changed columns) {
+	changed = columns{}
+	for _, cO := range old {
+		for _, cN := range new {
+			if cO.name == cN.name && cO.dataType != cN.dataType {
+				changed = append(changed, cN)
+			}
+		}
+	}
+	return
+}
+
+func (old columns) diffs(new columns) (removed columns, added columns, changed columns) {
+	return old.getNewIn(new), new.getNewIn(old), old.getChangedTypes(new)
+}
+
+func getColumn(from interface{}) (to column) {
+	c := from.(map[string]interface{})
+	return column{
+		name:     c["name"].(string),
+		dataType: c["type"].(string),
+	}
+}
+
+func getColumns(from interface{}) (to columns) {
+	cols := from.([]interface{})
+	to = make(columns, len(cols))
+	for i, c := range cols {
+		to[i] = getColumn(c)
+	}
+	return to
+}
+
 // CreateTable implements schema.CreateFunc
 func CreateTable(d *schema.ResourceData, meta interface{}) error {
 	db := meta.(*sql.DB)
@@ -136,17 +205,8 @@ func CreateTable(d *schema.ResourceData, meta interface{}) error {
 	schema := d.Get("schema").(string)
 	name := d.Get("name").(string)
 
-	// This type conversion is due to the test framework in the terraform-plugin-sdk having limited support
-	// for data types in the HCL2ValueFromConfigValue method.
-	columns := []snowflake.Column{}
-
-	for _, column := range d.Get("column").([]interface{}) {
-		typed := column.(map[string]interface{})
-		columnDef := snowflake.Column{}
-		columnDef.WithName(typed["name"].(string)).WithType(typed["type"].(string))
-		columns = append(columns, columnDef)
-	}
-	builder := snowflake.TableWithColumnDefinitions(name, database, schema, columns)
+	columns := getColumns(d.Get("column").([]interface{}))
+	builder := snowflake.TableWithColumnDefinitions(name, database, schema, columns.toSnowflakeColumns())
 
 	// Set optionals
 	if v, ok := d.GetOk("comment"); ok {
@@ -246,6 +306,31 @@ func UpdateTable(d *schema.ResourceData, meta interface{}) error {
 			return errors.Wrapf(err, "error updating table comment on %v", d.Id())
 		}
 	}
+	if d.HasChange("column") {
+		old, new := d.GetChange("column")
+		removed, added, changed := getColumns(old).diffs(getColumns(new))
+		for _, cA := range removed {
+			q := builder.DropColumn(cA.name)
+			err := snowflake.Exec(db, q)
+			if err != nil {
+				return errors.Wrapf(err, "error dropping column on %v", d.Id())
+			}
+		}
+		for _, cA := range added {
+			q := builder.AddColumn(cA.name, cA.dataType)
+			err := snowflake.Exec(db, q)
+			if err != nil {
+				return errors.Wrapf(err, "error adding column on %v", d.Id())
+			}
+		}
+		for _, cA := range changed {
+			q := builder.ChangeColumnType(cA.name, cA.dataType)
+			err := snowflake.Exec(db, q)
+			if err != nil {
+				return errors.Wrapf(err, "error changing column type on %v", d.Id())
+			}
+		}
+	}
 
 	return ReadTable(d, meta)
 }
diff --git a/pkg/snowflake/table.go b/pkg/snowflake/table.go
index a9da0c74..6d6b1063 100644
--- a/pkg/snowflake/table.go
+++ b/pkg/snowflake/table.go
@@ -160,6 +160,29 @@ func (tb *TableBuilder) ChangeComment(c string) string {
 	return fmt.Sprintf(`ALTER TABLE %v SET COMMENT = '%v'`, tb.QualifiedName(), EscapeString(c))
 }
 
+// AddColumn returns the SQL query that will add a new column to the table.
+func (tb *TableBuilder) AddColumn(name string, dataType string) string {
+	col := Column{
+		name:  name,
+		_type: dataType,
+	}
+	return fmt.Sprintf(`ALTER TABLE %s ADD COLUMN %s`, tb.QualifiedName(), col.getColumnDefinition())
+}
+
+// DropColumn returns the SQL query that will add a new column to the table.
+func (tb *TableBuilder) DropColumn(name string) string {
+	return fmt.Sprintf(`ALTER TABLE %s DROP COLUMN "%s"`, tb.QualifiedName(), name)
+}
+
+// ChangeColumnType returns the SQL query that will change the type of the named column to the given type.
+func (tb *TableBuilder) ChangeColumnType(name string, dataType string) string {
+	col := Column{
+		name:  name,
+		_type: dataType,
+	}
+	return fmt.Sprintf(`ALTER TABLE %s MODIFY COLUMN %s`, tb.QualifiedName(), col.getColumnDefinition())
+}
+
 // RemoveComment returns the SQL query that will remove the comment on the table.
 func (tb *TableBuilder) RemoveComment() string {
 	return fmt.Sprintf(`ALTER TABLE %v UNSET COMMENT`, tb.QualifiedName())


Test patch:
diff --git a/pkg/resources/table_acceptance_test.go b/pkg/resources/table_acceptance_test.go
index f3e4bf90..74bbc3ed 100644
--- a/pkg/resources/table_acceptance_test.go
+++ b/pkg/resources/table_acceptance_test.go
@@ -22,6 +22,25 @@ func TestAcc_Table(t *testing.T) {
 					resource.TestCheckResourceAttr("snowflake_table.test_table", "database", accName),
 					resource.TestCheckResourceAttr("snowflake_table.test_table", "schema", accName),
 					resource.TestCheckResourceAttr("snowflake_table.test_table", "comment", "Terraform acceptance test"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table", "column.#", "2"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table", "column.0.name", "column1"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table", "column.0.type", "VARIANT"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table", "column.1.name", "column2"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table", "column.1.type", "VARCHAR(16)"),
+				),
+			},
+			{
+				Config: tableConfig2(accName),
+				Check: resource.ComposeTestCheckFunc(
+					resource.TestCheckResourceAttr("snowflake_table.test_table", "name", accName),
+					resource.TestCheckResourceAttr("snowflake_table.test_table", "database", accName),
+					resource.TestCheckResourceAttr("snowflake_table.test_table", "schema", accName),
+					resource.TestCheckResourceAttr("snowflake_table.test_table", "comment", "Terraform acceptance test"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table", "column.#", "2"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table", "column.0.name", "column2"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table", "column.0.type", "VARCHAR(16777216)"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table", "column.1.name", "column3"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table", "column.1.type", "FLOAT"),
 				),
 			},
 		},
@@ -50,10 +69,41 @@ resource "snowflake_table" "test_table" {
 		name = "column1"
 		type = "VARIANT"
 	}
+	column {
+		name = "column2"
+		type = "VARCHAR(16)"
+	}
+}
+`
+	return fmt.Sprintf(s, name, name, name)
+}
+
+func tableConfig2(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_table" {
+	database = snowflake_database.test_database.name
+	schema   = snowflake_schema.test_schema.name
+	name     = "%s"
+	comment  = "Terraform acceptance test"
 	column {
 		name = "column2"
 		type = "VARCHAR(16777216)"
 	}
+	column {
+		name = "column3"
+		type = "FLOAT"
+	}
 }
 `
 	return fmt.Sprintf(s, name, name, name)
diff --git a/pkg/snowflake/table_test.go b/pkg/snowflake/table_test.go
index 010562e6..8a17e401 100644
--- a/pkg/snowflake/table_test.go
+++ b/pkg/snowflake/table_test.go
@@ -41,6 +41,24 @@ func TestTableRemoveComment(t *testing.T) {
 	r.Equal(s.RemoveComment(), `ALTER TABLE "test_db"."test_schema"."test_table" UNSET COMMENT`)
 }
 
+func TestTableAddColumn(t *testing.T) {
+	r := require.New(t)
+	s := Table("test_table", "test_db", "test_schema")
+	r.Equal(s.AddColumn("new_column", "VARIANT"), `ALTER TABLE "test_db"."test_schema"."test_table" ADD COLUMN "new_column" VARIANT`)
+}
+
+func TestTableDropColumn(t *testing.T) {
+	r := require.New(t)
+	s := Table("test_table", "test_db", "test_schema")
+	r.Equal(s.DropColumn("old_column"), `ALTER TABLE "test_db"."test_schema"."test_table" DROP COLUMN "old_column"`)
+}
+
+func TestTableChangeColumnType(t *testing.T) {
+	r := require.New(t)
+	s := Table("test_table", "test_db", "test_schema")
+	r.Equal(s.ChangeColumnType("old_column", "BIGINT"), `ALTER TABLE "test_db"."test_schema"."test_table" MODIFY COLUMN "old_column" BIGINT`)
+}
+
 func TestTableDrop(t *testing.T) {
 	r := require.New(t)
 	s := Table("test_table", "test_db", "test_schema")