feat: Identity Column Support (#726)

<!-- Feel free to delete comments as you fill this in -->
Adds support for a column identity/autoincrement for snowflake tables.

Only supports adding an identity to a column for the follow operations
* `CREATE TABLE`
* `ALTER TABLE ADD COLUMN`

<!-- summary of changes -->

## Test Plan
<!-- detail ways in which this PR has been tested or needs to be tested -->
* [x] acceptance tests
* [x] unit tests

## References
* https://docs.snowflake.com/en/sql-reference/sql/create-table.html#optional-parameters (see autoincrement/identity section)
* #538

docs/resources/table.md

@@ -44,6 +44,17 @@ resource "snowflake_table" "table" {
     }
   }
 
+  column {
+    name     = "identity"
+    type     = "NUMBER(38,0)"
+    nullable = true
+
+    identity {
+      start_num = 1
+      step_num  = 3
+    }
+  }
+
   column {
     name     = "data"
     type     = "text"
@@ -104,6 +115,7 @@ Optional:
 
 - **comment** (String) Column comment
 - **default** (Block List, Max: 1) Defines the column default value; note due to limitations of Snowflake's ALTER TABLE ADD/MODIFY COLUMN updates to default will not be applied (see [below for nested schema](#nestedblock--column--default))
+- **identity** (Block List, Max: 1) Defines the identity start/step values for a column. **Note** Identity/default are mutually exclusive. (see [below for nested schema](#nestedblock--column--identity))
 - **nullable** (Boolean) Whether this column can contain null values. **Note**: Depending on your Snowflake version, the default value will not suffice if this column is used in a primary key constraint.
 
 <a id="nestedblock--column--default"></a>
@@ -116,6 +128,15 @@ Optional:
 - **sequence** (String) The default sequence to use for the column
 
 
+<a id="nestedblock--column--identity"></a>
+### Nested Schema for `column.identity`
+
+Optional:
+
+- **start_num** (Number) The number to start incrementing at.
+- **step_num** (Number) Step size to increment by.
+
+
 
 <a id="nestedblock--primary_key"></a>
 ### Nested Schema for `primary_key`

examples/resources/snowflake_table/resource.tf

@@ -29,6 +29,17 @@ resource "snowflake_table" "table" {
     }
   }
 
+  column {
+    name     = "identity"
+    type     = "NUMBER(38,0)"
+    nullable = true
+
+    identity {
+      start_num = 1
+      step_num  = 3
+    }
+  }
+
   column {
     name     = "data"
     type     = "text"

pkg/resources/table.go

@@ -95,6 +95,32 @@ var tableSchema = map[string]*schema.Schema{
 						},
 					},
 				},
+				/*Note: Identity and default are mutually exclusive. From what I can tell we can't enforce this here
+				the snowflake query will error so we can defer enforcement to there.
+				*/
+				"identity": {
+					Type:        schema.TypeList,
+					Optional:    true,
+					Description: "Defines the identity start/step values for a column. **Note** Identity/default are mutually exclusive.",
+					MinItems:    1,
+					MaxItems:    1,
+					Elem: &schema.Resource{
+						Schema: map[string]*schema.Schema{
+							"start_num": {
+								Type:        schema.TypeInt,
+								Optional:    true,
+								Description: "The number to start incrementing at.",
+								Default:     1,
+							},
+							"step_num": {
+								Type:        schema.TypeInt,
+								Optional:    true,
+								Description: "Step size to increment by.",
+								Default:     1,
+							},
+						},
+					},
+				},
 				"comment": {
 					Type:        schema.TypeString,
 					Optional:    true,
@@ -251,11 +277,23 @@ func (cd *columnDefault) _type() string {
 	return "unknown"
 }
 
+type columnIdentity struct {
+	startNum int
+	stepNum  int
+}
+
+func (identity *columnIdentity) toSnowflakeColumnIdentity() *snowflake.ColumnIdentity {
+	snowIdentity := snowflake.ColumnIdentity{}
+	return snowIdentity.WithStartNum(identity.startNum).WithStep(identity.stepNum)
+
+}
+
 type column struct {
 	name     string
 	dataType string
 	nullable bool
 	_default *columnDefault
+	identity *columnIdentity
 	comment  string
 }
 
@@ -266,6 +304,10 @@ func (c column) toSnowflakeColumn() snowflake.Column {
 		sC = sC.WithDefault(c._default.toSnowflakeColumnDefault())
 	}
 
+	if c.identity != nil {
+		sC = sC.WithIdentity(c.identity.toSnowflakeColumnIdentity())
+	}
+
 	return *sC.WithName(c.name).
 		WithType(c.dataType).
 		WithNullable(c.nullable).
@@ -349,20 +391,38 @@ func getColumnDefault(def map[string]interface{}) *columnDefault {
 	return nil
 }
 
+func getColumnIdentity(identity map[string]interface{}) *columnIdentity {
+	if len(identity) > 0 {
+
+		startNum := identity["start_num"].(int)
+		stepNum := identity["step_num"].(int)
+		return &columnIdentity{startNum, stepNum}
+	}
+
+	return nil
+}
+
 func getColumn(from interface{}) (to column) {
 	c := from.(map[string]interface{})
 	var cd *columnDefault
+	var id *columnIdentity
 
 	_default := c["default"].([]interface{})
+	identity := c["identity"].([]interface{})
+
 	if len(_default) == 1 {
 		cd = getColumnDefault(_default[0].(map[string]interface{}))
 	}
+	if len(identity) == 1 {
+		id = getColumnIdentity(identity[0].(map[string]interface{}))
+	}
 
 	return column{
 		name:     c["name"].(string),
 		dataType: c["type"].(string),
 		nullable: c["nullable"].(bool),
 		_default: cd,
+		identity: id,
 		comment:  c["comment"].(string),
 	}
 }
@@ -573,14 +633,17 @@ func UpdateTable(d *schema.ResourceData, meta interface{}) error {
 		}
 		for _, cA := range added {
 			var q string
-			if cA._default == nil {
-				q = builder.AddColumn(cA.name, cA.dataType, cA.nullable, nil, cA.comment)
+
+			if cA.identity == nil && cA._default == nil {
+				q = builder.AddColumn(cA.name, cA.dataType, cA.nullable, nil, nil, cA.comment)
+			} else if cA.identity != nil {
+				q = builder.AddColumn(cA.name, cA.dataType, cA.nullable, nil, cA.identity.toSnowflakeColumnIdentity(), cA.comment)
 			} else {
 				if cA._default._type() != "constant" {
 					return fmt.Errorf("Failed to add column %v => Only adding a column as a constant is supported by Snowflake", cA.name)
 				}
 
-				q = builder.AddColumn(cA.name, cA.dataType, cA.nullable, cA._default.toSnowflakeColumnDefault(), cA.comment)
+				q = builder.AddColumn(cA.name, cA.dataType, cA.nullable, cA._default.toSnowflakeColumnDefault(), nil, cA.comment)
 			}
 
 			err := snowflake.Exec(db, q)

pkg/resources/table_acceptance_test.go

@@ -1052,3 +1052,159 @@ resource "snowflake_table" "test_table" {
 `
 	return fmt.Sprintf(s, name, tagName, tag2Name)
 }
+
+func TestAcc_TableIdentity(t *testing.T) {
+	accName := strings.ToUpper(acctest.RandStringFromCharSet(10, acctest.CharSetAlpha))
+
+	resource.ParallelTest(t, resource.TestCase{
+		Providers: providers(),
+		Steps: []resource.TestStep{
+			{
+				Config: tableColumnWithIdentityDefault(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", "data_retention_days", "1"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table", "change_tracking", "false"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table", "comment", "Terraform acceptance test"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table", "column.#", "3"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table", "column.0.name", "column1"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table", "column.0.type", "NUMBER(38,0)"),
+					resource.TestCheckNoResourceAttr("snowflake_table.test_table", "column.0.type.default.0.expression"),
+					resource.TestCheckNoResourceAttr("snowflake_table.test_table", "column.0.type.default.0.sequence"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table", "column.1.name", "column2"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table", "column.1.type", "TIMESTAMP_NTZ(9)"),
+					resource.TestCheckNoResourceAttr("snowflake_table.test_table", "column.1.type.default.0.constant"),
+					resource.TestCheckNoResourceAttr("snowflake_table.test_table", "column.1.type.default.0.sequence"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table", "column.2.name", "column3"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table", "column.2.type", "NUMBER(38,0)"),
+					resource.TestCheckNoResourceAttr("snowflake_table.test_table", "column.2.type.default.0.constant"),
+					resource.TestCheckNoResourceAttr("snowflake_table.test_table", "column.2.type.default.0.expression"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table", "column.2.identity.0.start_num", "1"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table", "column.2.identity.0.step_num", "1"),
+					resource.TestCheckNoResourceAttr("snowflake_table.test_table", "primary_key"),
+				),
+			},
+			{
+				Config: tableColumnWithIdentity(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", "data_retention_days", "1"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table", "change_tracking", "false"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table", "comment", "Terraform acceptance test"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table", "column.#", "3"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table", "column.0.name", "column1"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table", "column.0.type", "NUMBER(38,0)"),
+					resource.TestCheckNoResourceAttr("snowflake_table.test_table", "column.0.type.default.0.expression"),
+					resource.TestCheckNoResourceAttr("snowflake_table.test_table", "column.0.type.default.0.sequence"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table", "column.1.name", "column2"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table", "column.1.type", "TIMESTAMP_NTZ(9)"),
+					resource.TestCheckNoResourceAttr("snowflake_table.test_table", "column.1.type.default.0.constant"),
+					resource.TestCheckNoResourceAttr("snowflake_table.test_table", "column.1.type.default.0.sequence"),
+					resource.TestCheckNoResourceAttr("snowflake_table.test_table", "column.2.type.default.0.constant"),
+					resource.TestCheckNoResourceAttr("snowflake_table.test_table", "column.2.type.default.0.expression"),
+					resource.TestCheckNoResourceAttr("snowflake_table.test_table", "column.0.identity.0.start_num"),
+					resource.TestCheckNoResourceAttr("snowflake_table.test_table", "column.0.identity.0.step_num"),
+					// we've dropped the previous identity column and making sure that adding a new column as an identity works
+					resource.TestCheckResourceAttr("snowflake_table.test_table", "column.2.identity.0.start_num", "2"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table", "column.2.identity.0.step_num", "4"),
+					resource.TestCheckNoResourceAttr("snowflake_table.test_table", "primary_key"),
+				),
+			},
+		},
+	})
+}
+
+func tableColumnWithIdentityDefault(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_sequence" "test_seq" {
+	database            = snowflake_database.test_database.name
+	schema              = snowflake_schema.test_schema.name
+	name                = "%s"
+}
+
+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 = "column1"
+		type = "NUMBER(38,0)"
+	}
+	column {
+		name = "column2"
+		type = "TIMESTAMP_NTZ(9)"
+	}
+	column {
+		name = "column3"
+		type = "NUMBER(38,0)"
+		identity {
+		}
+	}
+}
+`
+	return fmt.Sprintf(s, name, name, name, name)
+}
+
+func tableColumnWithIdentity(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_sequence" "test_seq" {
+	database            = snowflake_database.test_database.name
+	schema              = snowflake_schema.test_schema.name
+	name                = "%s"
+}
+
+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 = "column1"
+		type = "NUMBER(38,0)"
+	}
+	column {
+		name = "column2"
+		type = "TIMESTAMP_NTZ(9)"
+	}
+	
+	column {
+		name = "column4"
+		type = "NUMBER(38,0)"
+		identity {
+			start_num = 2
+			step_num = 4
+		}
+	}
+}
+`
+	return fmt.Sprintf(s, name, name, name, name)
+}