diff --git a/docs/resources/table.md b/docs/resources/table.md
index 9e0b74c305..729c19222a 100644
--- a/docs/resources/table.md
+++ b/docs/resources/table.md
@@ -13,13 +13,21 @@ description: |-
 ## Example Usage
 
 ```terraform
-resource snowflake_table table {
-  database   = "database"
-  schema     = "schmea"
-  name       = "table"
-  comment    = "A table."
-  cluster_by = ["to_date(DATE)"]
-  
+resource "snowflake_schema" "schema" {
+  database            = "database"
+  name                = "schema"
+  data_retention_days = 1
+}
+
+resource "snowflake_table" "table" {
+  database            = snowflake_schema.schema.database
+  schema              = snowflake_schema.schema.name
+  name                = "table"
+  comment             = "A table."
+  cluster_by          = ["to_date(DATE)"]
+  data_retention_days = snowflake_schema.schema.data_retention_days
+  change_tracking     = false
+
   column {
     name     = "id"
     type     = "int"
@@ -37,10 +45,15 @@ resource snowflake_table table {
     type = "TIMESTAMP_NTZ(9)"
   }
 
+  column {
+    name    = "extra"
+    type    = "VARIANT"
+    comment = "extra data"
+  }
+
   primary_key {
     name = "my_key"
     keys = ["data"]
-
   }
 }
 ```
@@ -57,8 +70,10 @@ resource snowflake_table table {
 
 ### Optional
 
+- **change_tracking** (Boolean) Specifies whether to enable change tracking on the table. Default false.
 - **cluster_by** (List of String) A list of one or more table columns/expressions to be used as clustering key(s) for the table
 - **comment** (String) Specifies a comment for the table.
+- **data_retention_days** (Number) Specifies the retention period for the table so that Time Travel actions (SELECT, CLONE, UNDROP) can be performed on historical data in the table. Default value is 1, if you wish to inherit the parent schema setting then pass in the schema attribute to this argument.
 - **id** (String) The ID of this resource.
 - **primary_key** (Block List, Max: 1) Definitions of primary key constraint to create on table (see [below for nested schema](#nestedblock--primary_key))
 
@@ -76,6 +91,7 @@ Required:
 
 Optional:
 
+- **comment** (String) Column comment
 - **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.
 
 
diff --git a/examples/resources/snowflake_table/resource.tf b/examples/resources/snowflake_table/resource.tf
index 0578d91ff6..653eb59019 100644
--- a/examples/resources/snowflake_table/resource.tf
+++ b/examples/resources/snowflake_table/resource.tf
@@ -1,10 +1,18 @@
-resource snowflake_table table {
-  database   = "database"
-  schema     = "schmea"
-  name       = "table"
-  comment    = "A table."
-  cluster_by = ["to_date(DATE)"]
-  
+resource "snowflake_schema" "schema" {
+  database            = "database"
+  name                = "schema"
+  data_retention_days = 1
+}
+
+resource "snowflake_table" "table" {
+  database            = snowflake_schema.schema.database
+  schema              = snowflake_schema.schema.name
+  name                = "table"
+  comment             = "A table."
+  cluster_by          = ["to_date(DATE)"]
+  data_retention_days = snowflake_schema.schema.data_retention_days
+  change_tracking     = false
+
   column {
     name     = "id"
     type     = "int"
@@ -22,9 +30,14 @@ resource snowflake_table table {
     type = "TIMESTAMP_NTZ(9)"
   }
 
+  column {
+    name    = "extra"
+    type    = "VARIANT"
+    comment = "extra data"
+  }
+
   primary_key {
     name = "my_key"
     keys = ["data"]
-
   }
 }
diff --git a/pkg/resources/stream_acceptance_test.go b/pkg/resources/stream_acceptance_test.go
index b2efcbbdac..b6745b3cf3 100644
--- a/pkg/resources/stream_acceptance_test.go
+++ b/pkg/resources/stream_acceptance_test.go
@@ -45,10 +45,12 @@ resource "snowflake_schema" "test_schema" {
 }
 
 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"
+	database        = snowflake_database.test_database.name
+	schema          = snowflake_schema.test_schema.name
+	name            = "STREAM_ON_TABLE"
+	comment         = "Terraform acceptance test"
+	change_tracking = true
+
 	column {
 		name = "column1"
 		type = "VARIANT"
diff --git a/pkg/resources/stream_grant_acceptance_test.go b/pkg/resources/stream_grant_acceptance_test.go
index 77694d2364..c0e97df25b 100644
--- a/pkg/resources/stream_grant_acceptance_test.go
+++ b/pkg/resources/stream_grant_acceptance_test.go
@@ -76,10 +76,12 @@ resource "snowflake_role" "test" {
   name = "{{.role_name}}"
 }
 resource "snowflake_table" "test" {
-	database = snowflake_database.test.name
-	schema   = snowflake_schema.test.name
-	name     = "{{ .table_name }}"
-	comment  = "Terraform acceptance test"
+	database        = snowflake_database.test.name
+	schema          = snowflake_schema.test.name
+	name            = "{{ .table_name }}"
+	change_tracking = true
+	comment         = "Terraform acceptance test"
+
 	column {
 		name = "column1"
 		type = "VARIANT"
diff --git a/pkg/resources/table.go b/pkg/resources/table.go
index f7a1897939..5d51aa8d6a 100644
--- a/pkg/resources/table.go
+++ b/pkg/resources/table.go
@@ -10,6 +10,7 @@ import (
 
 	"github.com/chanzuckerberg/terraform-provider-snowflake/pkg/snowflake"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/validation"
 	"github.com/pkg/errors"
 )
 
@@ -65,6 +66,12 @@ var tableSchema = map[string]*schema.Schema{
 					Default:     true,
 					Description: "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.",
 				},
+				"comment": {
+					Type:        schema.TypeString,
+					Optional:    true,
+					Default:     "",
+					Description: "Column comment",
+				},
 			},
 		},
 	},
@@ -101,6 +108,19 @@ var tableSchema = map[string]*schema.Schema{
 			},
 		},
 	},
+	"data_retention_days": {
+		Type:         schema.TypeInt,
+		Optional:     true,
+		Default:      1,
+		Description:  "Specifies the retention period for the table so that Time Travel actions (SELECT, CLONE, UNDROP) can be performed on historical data in the table. Default value is 1, if you wish to inherit the parent schema setting then pass in the schema attribute to this argument.",
+		ValidateFunc: validation.IntBetween(0, 90),
+	},
+	"change_tracking": {
+		Type:        schema.TypeBool,
+		Optional:    true,
+		Default:     false,
+		Description: "Specifies whether to enable change tracking on the table. Default false.",
+	},
 }
 
 func Table() *schema.Resource {
@@ -167,11 +187,12 @@ type column struct {
 	name     string
 	dataType string
 	nullable bool
+	comment  string
 }
 
 func (c column) toSnowflakeColumn() snowflake.Column {
 	sC := snowflake.Column{}
-	return *sC.WithName(c.name).WithType(c.dataType).WithNullable(c.nullable)
+	return *sC.WithName(c.name).WithType(c.dataType).WithNullable(c.nullable).WithComment(c.comment)
 }
 
 type columns []column
@@ -207,19 +228,23 @@ type changedColumn struct {
 	newColumn             column //our new column
 	changedDataType       bool
 	changedNullConstraint bool
+	changedComment        bool
 }
 
 func (old columns) getChangedColumnProperties(new columns) (changed changedColumns) {
 	changed = changedColumns{}
 	for _, cO := range old {
 		for _, cN := range new {
-			changeColumn := changedColumn{cN, false, false}
+			changeColumn := changedColumn{cN, false, false, false}
 			if cO.name == cN.name && cO.dataType != cN.dataType {
 				changeColumn.changedDataType = true
 			}
 			if cO.name == cN.name && cO.nullable != cN.nullable {
 				changeColumn.changedNullConstraint = true
 			}
+			if cO.name == cN.name && cO.comment != cN.comment {
+				changeColumn.changedComment = true
+			}
 
 			changed = append(changed, changeColumn)
 		}
@@ -237,6 +262,7 @@ func getColumn(from interface{}) (to column) {
 		name:     c["name"].(string),
 		dataType: c["type"].(string),
 		nullable: c["nullable"].(bool),
+		comment:  c["comment"].(string),
 	}
 }
 
@@ -297,6 +323,14 @@ func CreateTable(d *schema.ResourceData, meta interface{}) error {
 		builder.WithPrimaryKey(pk.toSnowflakePrimaryKey())
 	}
 
+	if v, ok := d.GetOk("data_retention_days"); ok {
+		builder.WithDataRetentionTimeInDays(v.(int))
+	}
+
+	if v, ok := d.GetOk("change_tracking"); ok {
+		builder.WithChangeTracking(v.(bool))
+	}
+
 	stmt := builder.Create()
 	err := snowflake.Exec(db, stmt)
 	if err != nil {
@@ -361,14 +395,16 @@ func ReadTable(d *schema.ResourceData, meta interface{}) error {
 
 	// Set the relevant data in the state
 	toSet := map[string]interface{}{
-		"name":        table.TableName.String,
-		"owner":       table.Owner.String,
-		"database":    tableID.DatabaseName,
-		"schema":      tableID.SchemaName,
-		"comment":     table.Comment.String,
-		"column":      snowflake.NewColumns(tableDescription).Flatten(),
-		"cluster_by":  snowflake.ClusterStatementToList(table.ClusterBy.String),
-		"primary_key": snowflake.FlattenTablePrimaryKey(pkDescription),
+		"name":                table.TableName.String,
+		"owner":               table.Owner.String,
+		"database":            tableID.DatabaseName,
+		"schema":              tableID.SchemaName,
+		"comment":             table.Comment.String,
+		"column":              snowflake.NewColumns(tableDescription).Flatten(),
+		"cluster_by":          snowflake.ClusterStatementToList(table.ClusterBy.String),
+		"primary_key":         snowflake.FlattenTablePrimaryKey(pkDescription),
+		"data_retention_days": table.RetentionTime.Int32,
+		"change_tracking":     (table.ChangeTracking.String == "ON"),
 	}
 
 	for key, val := range toSet {
@@ -430,7 +466,7 @@ func UpdateTable(d *schema.ResourceData, meta interface{}) error {
 			}
 		}
 		for _, cA := range added {
-			q := builder.AddColumn(cA.name, cA.dataType, cA.nullable)
+			q := builder.AddColumn(cA.name, cA.dataType, cA.nullable, cA.comment)
 			err := snowflake.Exec(db, q)
 			if err != nil {
 				return errors.Wrapf(err, "error adding column on %v", d.Id())
@@ -456,6 +492,14 @@ func UpdateTable(d *schema.ResourceData, meta interface{}) error {
 
 				}
 			}
+			if cA.changedComment {
+				q := builder.ChangeColumnComment(cA.newColumn.name, cA.newColumn.comment)
+				err := snowflake.Exec(db, q)
+				if err != nil {
+					return errors.Wrapf(err, "error changing property on %v", d.Id())
+
+				}
+			}
 
 		}
 	}
@@ -483,6 +527,24 @@ func UpdateTable(d *schema.ResourceData, meta interface{}) error {
 			}
 		}
 	}
+	if d.HasChange("data_retention_days") {
+		_, ndr := d.GetChange("data_retention_days")
+
+		q := builder.ChangeDataRetention(ndr.(int))
+		err := snowflake.Exec(db, q)
+		if err != nil {
+			return errors.Wrapf(err, "error changing property on %v", d.Id())
+		}
+	}
+	if d.HasChange("change_tracking") {
+		_, nct := d.GetChange("change_tracking")
+
+		q := builder.ChangeChangeTracking(nct.(bool))
+		err := snowflake.Exec(db, q)
+		if err != nil {
+			return errors.Wrapf(err, "error changing property on %v", d.Id())
+		}
+	}
 
 	return ReadTable(d, meta)
 }
diff --git a/pkg/resources/table_acceptance_test.go b/pkg/resources/table_acceptance_test.go
index 4570b7a03d..dcb056d4dc 100644
--- a/pkg/resources/table_acceptance_test.go
+++ b/pkg/resources/table_acceptance_test.go
@@ -13,6 +13,7 @@ func TestAcc_Table(t *testing.T) {
 	accName := strings.ToUpper(acctest.RandStringFromCharSet(10, acctest.CharSetAlpha))
 
 	table2Name := strings.ToUpper(acctest.RandStringFromCharSet(10, acctest.CharSetAlpha))
+	table3Name := strings.ToUpper(acctest.RandStringFromCharSet(10, acctest.CharSetAlpha))
 
 	resource.ParallelTest(t, resource.TestCase{
 		Providers: providers(),
@@ -23,12 +24,14 @@ func TestAcc_Table(t *testing.T) {
 					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.#", "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)"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table", "column.1.comment", ""),
 					resource.TestCheckNoResourceAttr("snowflake_table.test_table", "primary_key"),
 				),
 			},
@@ -38,12 +41,16 @@ func TestAcc_Table(t *testing.T) {
 					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.#", "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.0.comment", ""),
 					resource.TestCheckResourceAttr("snowflake_table.test_table", "column.1.name", "column3"),
 					resource.TestCheckResourceAttr("snowflake_table.test_table", "column.1.type", "FLOAT"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table", "column.1.comment", ""),
 					resource.TestCheckNoResourceAttr("snowflake_table.test_table", "cluster_by"),
 					resource.TestCheckNoResourceAttr("snowflake_table.test_table", "primary_key"),
 				),
@@ -54,6 +61,8 @@ func TestAcc_Table(t *testing.T) {
 					resource.TestCheckResourceAttr("snowflake_table.test_table2", "name", table2Name),
 					resource.TestCheckResourceAttr("snowflake_table.test_table2", "database", accName),
 					resource.TestCheckResourceAttr("snowflake_table.test_table2", "schema", accName),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "data_retention_days", "1"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "change_tracking", "false"),
 					resource.TestCheckResourceAttr("snowflake_table.test_table2", "comment", "Terraform acceptance test"),
 					resource.TestCheckResourceAttr("snowflake_table.test_table2", "column.#", "2"),
 					resource.TestCheckResourceAttr("snowflake_table.test_table2", "column.0.name", "COL1"),
@@ -70,6 +79,8 @@ func TestAcc_Table(t *testing.T) {
 					resource.TestCheckResourceAttr("snowflake_table.test_table2", "name", table2Name),
 					resource.TestCheckResourceAttr("snowflake_table.test_table2", "database", accName),
 					resource.TestCheckResourceAttr("snowflake_table.test_table2", "schema", accName),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "data_retention_days", "1"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "change_tracking", "false"),
 					resource.TestCheckResourceAttr("snowflake_table.test_table2", "comment", "Terraform acceptance test"),
 					resource.TestCheckResourceAttr("snowflake_table.test_table2", "column.#", "2"),
 					resource.TestCheckResourceAttr("snowflake_table.test_table2", "column.0.name", "COL1"),
@@ -85,6 +96,8 @@ func TestAcc_Table(t *testing.T) {
 					resource.TestCheckResourceAttr("snowflake_table.test_table2", "name", table2Name),
 					resource.TestCheckResourceAttr("snowflake_table.test_table2", "database", accName),
 					resource.TestCheckResourceAttr("snowflake_table.test_table2", "schema", accName),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "data_retention_days", "1"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "change_tracking", "false"),
 					resource.TestCheckResourceAttr("snowflake_table.test_table2", "comment", "Terraform acceptance test"),
 					resource.TestCheckResourceAttr("snowflake_table.test_table2", "column.#", "2"),
 					resource.TestCheckResourceAttr("snowflake_table.test_table2", "column.0.name", "COL1"),
@@ -100,6 +113,8 @@ func TestAcc_Table(t *testing.T) {
 					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.#", "2"),
 					resource.TestCheckResourceAttr("snowflake_table.test_table", "column.0.name", "column2"),
@@ -118,6 +133,8 @@ func TestAcc_Table(t *testing.T) {
 					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.#", "2"),
 					resource.TestCheckResourceAttr("snowflake_table.test_table", "column.0.name", "column2"),
@@ -137,6 +154,8 @@ func TestAcc_Table(t *testing.T) {
 					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.#", "2"),
 					resource.TestCheckResourceAttr("snowflake_table.test_table", "column.0.name", "column2"),
@@ -151,6 +170,148 @@ func TestAcc_Table(t *testing.T) {
 					resource.TestCheckResourceAttr("snowflake_table.test_table", "primary_key.0.name", "new_name"),
 				),
 			},
+			{
+				Config: tableConfig9CreateTableWithColumnComment(accName, table2Name),
+				Check: resource.ComposeTestCheckFunc(
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "name", table2Name),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "database", accName),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "schema", accName),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "data_retention_days", "1"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "change_tracking", "false"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "comment", "Terraform acceptance test"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "column.#", "2"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "column.0.name", "COL1"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "column.0.type", "VARCHAR(16777216)"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "column.0.nullable", "true"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "column.0.comment", ""),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "column.1.name", "COL2"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "column.1.type", "FLOAT"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "column.1.nullable", "false"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "column.1.comment", "some comment"),
+					resource.TestCheckNoResourceAttr("snowflake_table.test_table2", "cluster_by"),
+					resource.TestCheckNoResourceAttr("snowflake_table.test_table2", "primary_key"),
+				),
+			},
+			{
+				Config: tableConfig10AlterTableColumnComment(accName, table2Name),
+				Check: resource.ComposeTestCheckFunc(
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "name", table2Name),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "database", accName),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "schema", accName),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "data_retention_days", "1"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "change_tracking", "false"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "comment", "Terraform acceptance test"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "column.#", "2"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "column.0.name", "COL1"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "column.0.type", "VARCHAR(16777216)"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "column.0.nullable", "true"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "column.0.comment", "other comment"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "column.1.name", "COL2"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "column.1.type", "FLOAT"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "column.1.nullable", "false"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "column.1.comment", ""),
+					resource.TestCheckNoResourceAttr("snowflake_table.test_table2", "cluster_by"),
+					resource.TestCheckNoResourceAttr("snowflake_table.test_table2", "primary_key"),
+				),
+			},
+			{
+				Config: tableConfig11AlterTableAddColumnWithComment(accName, table2Name),
+				Check: resource.ComposeTestCheckFunc(
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "name", table2Name),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "database", accName),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "schema", accName),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "data_retention_days", "1"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "change_tracking", "false"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "comment", "Terraform acceptance test"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "column.#", "3"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "column.0.name", "COL1"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "column.0.type", "VARCHAR(16777216)"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "column.0.nullable", "true"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "column.0.comment", "other comment"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "column.1.name", "COL2"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "column.1.type", "FLOAT"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "column.1.nullable", "false"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "column.1.comment", ""),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "column.2.name", "COL3"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "column.2.type", "NUMBER(38,0)"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "column.2.nullable", "false"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table2", "column.2.comment", "extra"),
+					resource.TestCheckNoResourceAttr("snowflake_table.test_table2", "cluster_by"),
+					resource.TestCheckNoResourceAttr("snowflake_table.test_table2", "primary_key"),
+				),
+			},
+			{
+				Config: tableConfig12CreateTableWithDataRetention(accName, table3Name),
+				Check: resource.ComposeTestCheckFunc(
+					resource.TestCheckResourceAttr("snowflake_table.test_table3", "name", table3Name),
+					resource.TestCheckResourceAttr("snowflake_table.test_table3", "database", accName),
+					resource.TestCheckResourceAttr("snowflake_table.test_table3", "schema", accName),
+					resource.TestCheckResourceAttr("snowflake_table.test_table3", "data_retention_days", "10"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table3", "change_tracking", "false"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table3", "comment", "Terraform acceptance test"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table3", "column.#", "2"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table3", "column.0.name", "column1"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table3", "column.0.type", "VARIANT"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table3", "column.1.name", "column2"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table3", "column.1.type", "VARCHAR(16)"),
+					resource.TestCheckNoResourceAttr("snowflake_table.test_table3", "cluster_by"),
+					resource.TestCheckNoResourceAttr("snowflake_table.test_table3", "primary_key"),
+				),
+			},
+			{
+				Config: tableConfig13AlterTableDataRetention(accName, table3Name),
+				Check: resource.ComposeTestCheckFunc(
+					resource.TestCheckResourceAttr("snowflake_table.test_table3", "name", table3Name),
+					resource.TestCheckResourceAttr("snowflake_table.test_table3", "database", accName),
+					resource.TestCheckResourceAttr("snowflake_table.test_table3", "schema", accName),
+					resource.TestCheckResourceAttr("snowflake_table.test_table3", "data_retention_days", "0"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table3", "change_tracking", "false"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table3", "comment", "Terraform acceptance test"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table3", "column.#", "2"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table3", "column.0.name", "column1"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table3", "column.0.type", "VARIANT"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table3", "column.1.name", "column2"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table3", "column.1.type", "VARCHAR(16)"),
+					resource.TestCheckNoResourceAttr("snowflake_table.test_table3", "cluster_by"),
+					resource.TestCheckNoResourceAttr("snowflake_table.test_table3", "primary_key"),
+				),
+			},
+			{
+				Config: tableConfig14AlterTableEnableChangeTracking(accName, table3Name),
+				Check: resource.ComposeTestCheckFunc(
+					resource.TestCheckResourceAttr("snowflake_table.test_table3", "name", table3Name),
+					resource.TestCheckResourceAttr("snowflake_table.test_table3", "database", accName),
+					resource.TestCheckResourceAttr("snowflake_table.test_table3", "schema", accName),
+					resource.TestCheckResourceAttr("snowflake_table.test_table3", "data_retention_days", "0"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table3", "change_tracking", "true"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table3", "comment", "Terraform acceptance test"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table3", "column.#", "2"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table3", "column.0.name", "column1"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table3", "column.0.type", "VARIANT"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table3", "column.1.name", "column2"),
+					resource.TestCheckResourceAttr("snowflake_table.test_table3", "column.1.type", "VARCHAR(16)"),
+					resource.TestCheckNoResourceAttr("snowflake_table.test_table3", "cluster_by"),
+					resource.TestCheckNoResourceAttr("snowflake_table.test_table3", "primary_key"),
+				),
+			},
+			{
+				Config: tableConfig15CreateTableWithChangeTracking(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", "true"),
+					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)"),
+					resource.TestCheckNoResourceAttr("snowflake_table.test_table", "cluster_by"),
+					resource.TestCheckNoResourceAttr("snowflake_table.test_table", "primary_key"),
+				),
+			},
 		},
 	})
 }
@@ -416,3 +577,237 @@ resource "snowflake_table" "test_table" {
 `
 	return fmt.Sprintf(s, name, name, name)
 }
+
+func tableConfig9CreateTableWithColumnComment(name string, table2Name 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_table2" {
+	database = snowflake_database.test_database.name
+	schema   = snowflake_schema.test_schema.name
+	name     = "%s"
+	comment  = "Terraform acceptance test"
+	column {
+		name = "COL1"
+		type = "VARCHAR(16777216)"
+	}
+	column {
+		name     = "COL2"
+		type     = "FLOAT"
+		nullable = false
+		comment  = "some comment"
+	}
+}
+`
+	return fmt.Sprintf(s, name, name, table2Name)
+}
+
+func tableConfig10AlterTableColumnComment(name string, table2Name 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_table2" {
+	database            = snowflake_database.test_database.name
+	schema              = snowflake_schema.test_schema.name
+	name                = "%s"
+	comment             = "Terraform acceptance test"
+	column {
+		name    = "COL1"
+		type    = "VARCHAR(16777216)"
+		comment = "other comment"
+	}
+	column {
+		name     = "COL2"
+		type     = "FLOAT"
+		nullable = false
+	}
+}
+`
+	return fmt.Sprintf(s, name, name, table2Name)
+}
+
+func tableConfig11AlterTableAddColumnWithComment(name string, table2Name 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_table2" {
+	database            = snowflake_database.test_database.name
+	schema              = snowflake_schema.test_schema.name
+	name                = "%s"
+	comment             = "Terraform acceptance test"
+	column {
+		name    = "COL1"
+		type    = "VARCHAR(16777216)"
+		comment = "other comment"
+	}
+	column {
+		name     = "COL2"
+		type     = "FLOAT"
+		nullable = false
+	}
+	column {
+		name     = "COL3"
+		type     = "NUMBER(38,0)"
+		nullable = false
+		comment  = "extra"
+	}
+}
+`
+	return fmt.Sprintf(s, name, name, table2Name)
+}
+
+func tableConfig12CreateTableWithDataRetention(name string, table3Name 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_table3" {
+	database            = snowflake_database.test_database.name
+	schema              = snowflake_schema.test_schema.name
+	name                = "%s"
+	comment             = "Terraform acceptance test"
+	data_retention_days = 10
+	column {
+		name = "column1"
+		type = "VARIANT"
+	}
+	column {
+		name = "column2"
+		type = "VARCHAR(16)"
+	}
+}
+`
+	return fmt.Sprintf(s, name, name, table3Name)
+}
+
+func tableConfig13AlterTableDataRetention(name string, table3Name 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_table3" {
+	database            = snowflake_database.test_database.name
+	schema              = snowflake_schema.test_schema.name
+	name                = "%s"
+	comment             = "Terraform acceptance test"
+	data_retention_days = 0
+	column {
+		name = "column1"
+		type = "VARIANT"
+	}
+	column {
+		name = "column2"
+		type = "VARCHAR(16)"
+	}
+}
+`
+	return fmt.Sprintf(s, name, name, table3Name)
+}
+
+func tableConfig14AlterTableEnableChangeTracking(name string, table3Name 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_table3" {
+	database            = snowflake_database.test_database.name
+	schema              = snowflake_schema.test_schema.name
+	name                = "%s"
+	comment             = "Terraform acceptance test"
+	data_retention_days = 0
+	change_tracking     = true
+	column {
+		name = "column1"
+		type = "VARIANT"
+	}
+	column {
+		name = "column2"
+		type = "VARCHAR(16)"
+	}
+}
+`
+	return fmt.Sprintf(s, name, name, table3Name)
+}
+
+func tableConfig15CreateTableWithChangeTracking(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"
+	change_tracking     = true
+	column {
+		name = "column1"
+		type = "VARIANT"
+	}
+	column {
+		name = "column2"
+		type = "VARCHAR(16)"
+	}
+}
+`
+	return fmt.Sprintf(s, name, name, name)
+}
diff --git a/pkg/resources/table_test.go b/pkg/resources/table_test.go
index ddb6f0dc52..dfd961e794 100644
--- a/pkg/resources/table_test.go
+++ b/pkg/resources/table_test.go
@@ -22,17 +22,32 @@ func TestTableCreate(t *testing.T) {
 	r := require.New(t)
 
 	in := map[string]interface{}{
-		"name":        "good_name",
-		"database":    "database_name",
-		"schema":      "schema_name",
-		"comment":     "great comment",
-		"column":      []interface{}{map[string]interface{}{"name": "column1", "type": "OBJECT"}, map[string]interface{}{"name": "column2", "type": "VARCHAR", "nullable": false}},
+		"name":     "good_name",
+		"database": "database_name",
+		"schema":   "schema_name",
+		"comment":  "great comment",
+		"column": []interface{}{
+			map[string]interface{}{
+				"name": "column1",
+				"type": "OBJECT",
+			},
+			map[string]interface{}{
+				"name":     "column2",
+				"type":     "VARCHAR",
+				"nullable": false,
+			},
+			map[string]interface{}{
+				"name":    "column3",
+				"type":    "INT",
+				"comment": "some comment",
+			},
+		},
 		"primary_key": []interface{}{map[string]interface{}{"name": "MY_KEY", "keys": []interface{}{"column1"}}},
 	}
 	d := table(t, "database_name|schema_name|good_name", in)
 
 	WithMockDb(t, func(db *sql.DB, mock sqlmock.Sqlmock) {
-		mock.ExpectExec(`CREATE TABLE "database_name"."schema_name"."good_name" \("column1" OBJECT, "column2" VARCHAR NOT NULL ,CONSTRAINT "MY_KEY" PRIMARY KEY\("column1"\)\) COMMENT = 'great comment'`).WillReturnResult(sqlmock.NewResult(1, 1))
+		mock.ExpectExec(`CREATE TABLE "database_name"."schema_name"."good_name" \("column1" OBJECT COMMENT '', "column2" VARCHAR NOT NULL COMMENT '', "column3" INT COMMENT 'some comment' ,CONSTRAINT "MY_KEY" PRIMARY KEY\("column1"\)\) COMMENT = 'great comment'`).WillReturnResult(sqlmock.NewResult(1, 1))
 		expectTableRead(mock)
 		err := resources.CreateTable(d, db)
 		r.NoError(err)
diff --git a/pkg/snowflake/table.go b/pkg/snowflake/table.go
index de3343634b..90e9748d30 100644
--- a/pkg/snowflake/table.go
+++ b/pkg/snowflake/table.go
@@ -12,54 +12,74 @@ import (
 	"github.com/pkg/errors"
 )
 
+// PrimaryKey structure that represents a tables primary key
 type PrimaryKey struct {
 	name string
 	keys []string
 }
 
+// WithName set the primary key name
 func (pk *PrimaryKey) WithName(name string) *PrimaryKey {
 	pk.name = name
 	return pk
 }
 
+// WithKeys set the primary key keys
 func (pk *PrimaryKey) WithKeys(keys []string) *PrimaryKey {
 	pk.keys = keys
 	return pk
 }
 
+// Column structure that represents a table column
 type Column struct {
 	name     string
 	_type    string // type is reserved
 	nullable bool
+	comment  string // pointer as value is nullable
 }
 
+// WithName set the column name
 func (c *Column) WithName(name string) *Column {
 	c.name = name
 	return c
 }
+
+// WithType set the column type
 func (c *Column) WithType(t string) *Column {
 	c._type = t
 	return c
 }
 
+// WithNullable set if the column is nullable
 func (c *Column) WithNullable(nullable bool) *Column {
 	c.nullable = nullable
 	return c
 }
 
-func (c *Column) getColumnDefinition(withInlineConstraints bool) string {
+// WithComment set the column comment
+func (c *Column) WithComment(comment string) *Column {
+	c.comment = comment
+	return c
+}
+
+func (c *Column) getColumnDefinition(withInlineConstraints bool, withComment bool) string {
 
 	if c == nil {
 		return ""
 	}
 	var colDef strings.Builder
 	colDef.WriteString(fmt.Sprintf(`"%v" %v`, EscapeString(c.name), EscapeString(c._type)))
+
 	if withInlineConstraints {
 		if !c.nullable {
 			colDef.WriteString(` NOT NULL`)
 		}
 	}
 
+	if withComment {
+		colDef.WriteString(fmt.Sprintf(` COMMENT '%v'`, EscapeString(c.comment)))
+	}
+
 	return colDef.String()
 
 }
@@ -117,6 +137,7 @@ func NewColumns(tds []tableDescription) Columns {
 			name:     td.Name.String,
 			_type:    td.Type.String,
 			nullable: td.IsNullable(),
+			comment:  td.Comment.String,
 		})
 	}
 	return Columns(cs)
@@ -129,17 +150,18 @@ func (c Columns) Flatten() []interface{} {
 		flat["name"] = col.name
 		flat["type"] = col._type
 		flat["nullable"] = col.nullable
+		flat["comment"] = col.comment
 
 		flattened = append(flattened, flat)
 	}
 	return flattened
 }
 
-func (c Columns) getColumnDefinitions(withInlineConstraints bool) string {
+func (c Columns) getColumnDefinitions(withInlineConstraints bool, withComments bool) string {
 	// TODO(el): verify Snowflake reflects column order back in desc table calls
 	columnDefinitions := []string{}
 	for _, column := range c {
-		columnDefinitions = append(columnDefinitions, column.getColumnDefinition(withInlineConstraints))
+		columnDefinitions = append(columnDefinitions, column.getColumnDefinition(withInlineConstraints, withComments))
 	}
 
 	// NOTE: intentionally blank leading space
@@ -148,13 +170,16 @@ func (c Columns) getColumnDefinitions(withInlineConstraints bool) string {
 
 // TableBuilder abstracts the creation of SQL queries for a Snowflake schema
 type TableBuilder struct {
-	name       string
-	db         string
-	schema     string
-	columns    Columns
-	comment    string
-	clusterBy  []string
-	primaryKey PrimaryKey
+	name                    string
+	db                      string
+	schema                  string
+	columns                 Columns
+	comment                 string
+	clusterBy               []string
+	primaryKey              PrimaryKey
+	dataRetentionTimeInDays int
+	changeTracking          bool
+	defaultDDLCollation     string
 }
 
 // QualifiedName prepends the db and schema if set and escapes everything nicely
@@ -196,11 +221,24 @@ func (tb *TableBuilder) WithClustering(c []string) *TableBuilder {
 	return tb
 }
 
+// WithPrimaryKey sets the primary key on the TableBuilder
 func (tb *TableBuilder) WithPrimaryKey(pk PrimaryKey) *TableBuilder {
 	tb.primaryKey = pk
 	return tb
 }
 
+// WithDataRetentionTimeInDays sets the data retention time on the TableBuilder
+func (tb *TableBuilder) WithDataRetentionTimeInDays(days int) *TableBuilder {
+	tb.dataRetentionTimeInDays = days
+	return tb
+}
+
+// WithChangeTracking sets the change tracking on the TableBuilder
+func (tb *TableBuilder) WithChangeTracking(changeTracking bool) *TableBuilder {
+	tb.changeTracking = changeTracking
+	return tb
+}
+
 //Function to get clustering definition
 func (tb *TableBuilder) GetClusterKeyString() string {
 
@@ -227,7 +265,7 @@ func quoteStringList(instrings []string) []string {
 func (tb *TableBuilder) getCreateStatementBody() string {
 	var q strings.Builder
 
-	colDef := tb.columns.getColumnDefinitions(true)
+	colDef := tb.columns.getColumnDefinitions(true, true)
 
 	if len(tb.primaryKey.keys) > 0 {
 		colDef = strings.TrimSuffix(colDef, ")") //strip trailing
@@ -313,6 +351,9 @@ func (tb *TableBuilder) Create() string {
 
 	}
 
+	q.WriteString(fmt.Sprintf(` DATA_RETENTION_TIME_IN_DAYS = %d`, tb.dataRetentionTimeInDays))
+	q.WriteString(fmt.Sprintf(` CHANGE_TRACKING = %t`, tb.changeTracking))
+
 	return q.String()
 }
 
@@ -326,14 +367,25 @@ func (tb *TableBuilder) ChangeComment(c string) string {
 	return fmt.Sprintf(`ALTER TABLE %v SET COMMENT = '%v'`, tb.QualifiedName(), EscapeString(c))
 }
 
+// ChangeDataRetention returns the SQL query that will update the DATA_RETENTION_TIME_IN_DAYS on the table
+func (tb *TableBuilder) ChangeDataRetention(days int) string {
+	return fmt.Sprintf(`ALTER TABLE %v SET DATA_RETENTION_TIME_IN_DAYS = %d`, tb.QualifiedName(), days)
+}
+
+// ChangeChangeTracking returns the SQL query that will update the CHANGE_TRACKING on the table
+func (tb *TableBuilder) ChangeChangeTracking(changeTracking bool) string {
+	return fmt.Sprintf(`ALTER TABLE %v SET CHANGE_TRACKING = %t`, tb.QualifiedName(), changeTracking)
+}
+
 // AddColumn returns the SQL query that will add a new column to the table.
-func (tb *TableBuilder) AddColumn(name string, dataType string, nullable bool) string {
+func (tb *TableBuilder) AddColumn(name string, dataType string, nullable bool, comment string) string {
 	col := Column{
 		name:     name,
 		_type:    dataType,
 		nullable: nullable,
+		comment:  comment,
 	}
-	return fmt.Sprintf(`ALTER TABLE %s ADD COLUMN %s`, tb.QualifiedName(), col.getColumnDefinition(true))
+	return fmt.Sprintf(`ALTER TABLE %s ADD COLUMN %s`, tb.QualifiedName(), col.getColumnDefinition(true, true))
 }
 
 // DropColumn returns the SQL query that will add a new column to the table.
@@ -347,7 +399,12 @@ func (tb *TableBuilder) ChangeColumnType(name string, dataType string) string {
 		name:  name,
 		_type: dataType,
 	}
-	return fmt.Sprintf(`ALTER TABLE %s MODIFY COLUMN %s`, tb.QualifiedName(), col.getColumnDefinition(false))
+
+	return fmt.Sprintf(`ALTER TABLE %s MODIFY COLUMN %s`, tb.QualifiedName(), col.getColumnDefinition(false, false))
+}
+
+func (tb *TableBuilder) ChangeColumnComment(name string, comment string) string {
+	return fmt.Sprintf(`ALTER TABLE %s MODIFY COLUMN "%v" COMMENT '%v'`, tb.QualifiedName(), EscapeString(name), EscapeString(comment))
 }
 
 // RemoveComment returns the SQL query that will remove the comment on the table.
@@ -411,7 +468,7 @@ type table struct {
 	Rows                sql.NullString `db:"row"`
 	Bytes               sql.NullString `db:"bytes"`
 	Owner               sql.NullString `db:"owner"`
-	RetentionTime       sql.NullString `db:"retention_time"`
+	RetentionTime       sql.NullInt32  `db:"retention_time"`
 	AutomaticClustering sql.NullString `db:"automatic_clustering"`
 	ChangeTracking      sql.NullString `db:"change_tracking"`
 }
@@ -427,6 +484,7 @@ type tableDescription struct {
 	Type     sql.NullString `db:"type"`
 	Kind     sql.NullString `db:"kind"`
 	Nullable sql.NullString `db:"null?"`
+	Comment  sql.NullString `db:"comment"`
 }
 
 func (td *tableDescription) IsNullable() bool {
diff --git a/pkg/snowflake/table_test.go b/pkg/snowflake/table_test.go
index fa2d1320a2..5e9c0c7045 100644
--- a/pkg/snowflake/table_test.go
+++ b/pkg/snowflake/table_test.go
@@ -19,22 +19,29 @@ func TestTableCreate(t *testing.T) {
 			name:     "column2",
 			_type:    "VARCHAR",
 			nullable: true,
+			comment:  "only populated when data is available",
 		},
 	}
 
 	s.WithColumns(Columns(cols))
 	r.Equal(s.QualifiedName(), `"test_db"."test_schema"."test_table"`)
 
-	r.Equal(`CREATE TABLE "test_db"."test_schema"."test_table" ("column1" OBJECT, "column2" VARCHAR)`, s.Create())
+	r.Equal(`CREATE TABLE "test_db"."test_schema"."test_table" ("column1" OBJECT COMMENT '', "column2" VARCHAR COMMENT 'only populated when data is available') DATA_RETENTION_TIME_IN_DAYS = 0 CHANGE_TRACKING = false`, s.Create())
 
 	s.WithComment("Test Comment")
-	r.Equal(s.Create(), `CREATE TABLE "test_db"."test_schema"."test_table" ("column1" OBJECT, "column2" VARCHAR) COMMENT = 'Test Comment'`)
+	r.Equal(s.Create(), `CREATE TABLE "test_db"."test_schema"."test_table" ("column1" OBJECT COMMENT '', "column2" VARCHAR COMMENT 'only populated when data is available') COMMENT = 'Test Comment' DATA_RETENTION_TIME_IN_DAYS = 0 CHANGE_TRACKING = false`)
 
 	s.WithClustering([]string{"column1"})
-	r.Equal(s.Create(), `CREATE TABLE "test_db"."test_schema"."test_table" ("column1" OBJECT, "column2" VARCHAR) COMMENT = 'Test Comment' CLUSTER BY LINEAR(column1)`)
+	r.Equal(s.Create(), `CREATE TABLE "test_db"."test_schema"."test_table" ("column1" OBJECT COMMENT '', "column2" VARCHAR COMMENT 'only populated when data is available') COMMENT = 'Test Comment' CLUSTER BY LINEAR(column1) DATA_RETENTION_TIME_IN_DAYS = 0 CHANGE_TRACKING = false`)
 
 	s.WithPrimaryKey(PrimaryKey{name: "MY_KEY", keys: []string{"column1"}})
-	r.Equal(s.Create(), `CREATE TABLE "test_db"."test_schema"."test_table" ("column1" OBJECT, "column2" VARCHAR ,CONSTRAINT "MY_KEY" PRIMARY KEY("column1")) COMMENT = 'Test Comment' CLUSTER BY LINEAR(column1)`)
+	r.Equal(s.Create(), `CREATE TABLE "test_db"."test_schema"."test_table" ("column1" OBJECT COMMENT '', "column2" VARCHAR COMMENT 'only populated when data is available' ,CONSTRAINT "MY_KEY" PRIMARY KEY("column1")) COMMENT = 'Test Comment' CLUSTER BY LINEAR(column1) DATA_RETENTION_TIME_IN_DAYS = 0 CHANGE_TRACKING = false`)
+
+	s.WithDataRetentionTimeInDays(10)
+	r.Equal(s.Create(), `CREATE TABLE "test_db"."test_schema"."test_table" ("column1" OBJECT COMMENT '', "column2" VARCHAR COMMENT 'only populated when data is available' ,CONSTRAINT "MY_KEY" PRIMARY KEY("column1")) COMMENT = 'Test Comment' CLUSTER BY LINEAR(column1) DATA_RETENTION_TIME_IN_DAYS = 10 CHANGE_TRACKING = false`)
+
+	s.WithChangeTracking(true)
+	r.Equal(s.Create(), `CREATE TABLE "test_db"."test_schema"."test_table" ("column1" OBJECT COMMENT '', "column2" VARCHAR COMMENT 'only populated when data is available' ,CONSTRAINT "MY_KEY" PRIMARY KEY("column1")) COMMENT = 'Test Comment' CLUSTER BY LINEAR(column1) DATA_RETENTION_TIME_IN_DAYS = 10 CHANGE_TRACKING = true`)
 }
 
 func TestTableChangeComment(t *testing.T) {
@@ -52,7 +59,13 @@ func TestTableRemoveComment(t *testing.T) {
 func TestTableAddColumn(t *testing.T) {
 	r := require.New(t)
 	s := Table("test_table", "test_db", "test_schema")
-	r.Equal(s.AddColumn("new_column", "VARIANT", true), `ALTER TABLE "test_db"."test_schema"."test_table" ADD COLUMN "new_column" VARIANT`)
+	r.Equal(s.AddColumn("new_column", "VARIANT", true, ""), `ALTER TABLE "test_db"."test_schema"."test_table" ADD COLUMN "new_column" VARIANT COMMENT ''`)
+}
+
+func TestTableAddColumnWithComment(t *testing.T) {
+	r := require.New(t)
+	s := Table("test_table", "test_db", "test_schema")
+	r.Equal(s.AddColumn("new_column", "VARIANT", true, "some comment"), `ALTER TABLE "test_db"."test_schema"."test_table" ADD COLUMN "new_column" VARIANT COMMENT 'some comment'`)
 }
 
 func TestTableDropColumn(t *testing.T) {
@@ -67,12 +80,30 @@ func TestTableChangeColumnType(t *testing.T) {
 	r.Equal(s.ChangeColumnType("old_column", "BIGINT"), `ALTER TABLE "test_db"."test_schema"."test_table" MODIFY COLUMN "old_column" BIGINT`)
 }
 
+func TestTableChangeColumnComment(t *testing.T) {
+	r := require.New(t)
+	s := Table("test_table", "test_db", "test_schema")
+	r.Equal(s.ChangeColumnComment("old_column", "some comment"), `ALTER TABLE "test_db"."test_schema"."test_table" MODIFY COLUMN "old_column" COMMENT 'some comment'`)
+}
+
 func TestTableChangeClusterBy(t *testing.T) {
 	r := require.New(t)
 	s := Table("test_table", "test_db", "test_schema")
 	r.Equal(s.ChangeClusterBy("column2, column3"), `ALTER TABLE "test_db"."test_schema"."test_table" CLUSTER BY LINEAR(column2, column3)`)
 }
 
+func TestTableChangeDataRetention(t *testing.T) {
+	r := require.New(t)
+	s := Table("test_table", "test_db", "test_schema")
+	r.Equal(s.ChangeDataRetention(5), `ALTER TABLE "test_db"."test_schema"."test_table" SET DATA_RETENTION_TIME_IN_DAYS = 5`)
+}
+
+func TestTableChangeChangeTracking(t *testing.T) {
+	r := require.New(t)
+	s := Table("test_table", "test_db", "test_schema")
+	r.Equal(s.ChangeChangeTracking(true), `ALTER TABLE "test_db"."test_schema"."test_table" SET CHANGE_TRACKING = true`)
+}
+
 func TestTableDropClusterBy(t *testing.T) {
 	r := require.New(t)
 	s := Table("test_table", "test_db", "test_schema")