diff --git a/clients/snowflake/ddl_test.go b/clients/snowflake/ddl_test.go
index f67e32fcf..c2e8a3e77 100644
--- a/clients/snowflake/ddl_test.go
+++ b/clients/snowflake/ddl_test.go
@@ -21,7 +21,7 @@ import (
 )
 
 func (s *SnowflakeTestSuite) TestMutateColumnsWithMemoryCacheDeletions() {
-	tableID := NewTableIdentifier("coffee_shop", "public", "orders", true)
+	tableID := NewTableIdentifier("coffee_shop", "public", "orders")
 
 	var cols columns.Columns
 	for colName, kindDetails := range map[string]typing.KindDetails{
@@ -51,7 +51,7 @@ func (s *SnowflakeTestSuite) TestMutateColumnsWithMemoryCacheDeletions() {
 }
 
 func (s *SnowflakeTestSuite) TestShouldDeleteColumn() {
-	tableID := NewTableIdentifier("coffee_shop", "orders", "public", true)
+	tableID := NewTableIdentifier("coffee_shop", "orders", "public")
 	var cols columns.Columns
 	for colName, kindDetails := range map[string]typing.KindDetails{
 		"id":          typing.Integer,
diff --git a/clients/snowflake/snowflake.go b/clients/snowflake/snowflake.go
index 1c3152310..9b2820be3 100644
--- a/clients/snowflake/snowflake.go
+++ b/clients/snowflake/snowflake.go
@@ -32,7 +32,7 @@ const (
 )
 
 func (s *Store) IdentifierFor(topicConfig kafkalib.TopicConfig, table string) types.TableIdentifier {
-	return NewTableIdentifier(topicConfig.Database, topicConfig.Schema, table, s.ShouldUppercaseEscapedNames())
+	return NewTableIdentifier(topicConfig.Database, topicConfig.Schema, table)
 }
 
 func (s *Store) GetTableConfig(tableData *optimization.TableData) (*types.DwhTableConfig, error) {
diff --git a/clients/snowflake/staging_test.go b/clients/snowflake/staging_test.go
index 8090c53f0..e743e1c4f 100644
--- a/clients/snowflake/staging_test.go
+++ b/clients/snowflake/staging_test.go
@@ -50,7 +50,7 @@ func (s *SnowflakeTestSuite) TestCastColValStaging() {
 }
 
 func (s *SnowflakeTestSuite) TestBackfillColumn() {
-	tableID := NewTableIdentifier("db", "public", "tableName", true)
+	tableID := NewTableIdentifier("db", "public", "tableName")
 
 	backfilledCol := columns.NewColumn("foo", typing.Boolean)
 	backfilledCol.SetDefaultValue(true)
@@ -130,7 +130,7 @@ func generateTableData(rows int) (TableIdentifier, *optimization.TableData) {
 		td.InsertRow(key, rowData, false)
 	}
 
-	return NewTableIdentifier("database", "schema", randomTableName, true), td
+	return NewTableIdentifier("database", "schema", randomTableName), td
 }
 
 func (s *SnowflakeTestSuite) TestPrepareTempTable() {
diff --git a/clients/snowflake/tableid.go b/clients/snowflake/tableid.go
index 9b08a47bd..eff9b8475 100644
--- a/clients/snowflake/tableid.go
+++ b/clients/snowflake/tableid.go
@@ -9,18 +9,16 @@ import (
 )
 
 type TableIdentifier struct {
-	database              string
-	schema                string
-	table                 string
-	uppercaseEscapedNames bool
+	database string
+	schema   string
+	table    string
 }
 
-func NewTableIdentifier(database, schema, table string, uppercaseEscapedNames bool) TableIdentifier {
+func NewTableIdentifier(database, schema, table string) TableIdentifier {
 	return TableIdentifier{
-		database:              database,
-		schema:                schema,
-		table:                 table,
-		uppercaseEscapedNames: uppercaseEscapedNames,
+		database: database,
+		schema:   schema,
+		table:    table,
 	}
 }
 
@@ -37,7 +35,7 @@ func (ti TableIdentifier) Table() string {
 }
 
 func (ti TableIdentifier) WithTable(table string) types.TableIdentifier {
-	return NewTableIdentifier(ti.database, ti.schema, table, ti.uppercaseEscapedNames)
+	return NewTableIdentifier(ti.database, ti.schema, table)
 }
 
 func (ti TableIdentifier) FullyQualifiedName() string {
@@ -45,6 +43,6 @@ func (ti TableIdentifier) FullyQualifiedName() string {
 		"%s.%s.%s",
 		ti.database,
 		ti.schema,
-		sql.EscapeNameIfNecessary(ti.table, ti.uppercaseEscapedNames, constants.Snowflake),
+		sql.EscapeNameIfNecessary(ti.table, true, constants.Snowflake),
 	)
 }
diff --git a/clients/snowflake/tableid_test.go b/clients/snowflake/tableid_test.go
index eed38849e..4f97313fa 100644
--- a/clients/snowflake/tableid_test.go
+++ b/clients/snowflake/tableid_test.go
@@ -7,22 +7,19 @@ import (
 )
 
 func TestTableIdentifier_WithTable(t *testing.T) {
-	tableID := NewTableIdentifier("database", "schema", "foo", true)
+	tableID := NewTableIdentifier("database", "schema", "foo")
 	tableID2 := tableID.WithTable("bar")
 	typedTableID2, ok := tableID2.(TableIdentifier)
 	assert.True(t, ok)
 	assert.Equal(t, "database", typedTableID2.Database())
 	assert.Equal(t, "schema", typedTableID2.Schema())
 	assert.Equal(t, "bar", tableID2.Table())
-	assert.True(t, typedTableID2.uppercaseEscapedNames)
 }
 
 func TestTableIdentifier_FullyQualifiedName(t *testing.T) {
-	// Table name that does not need escaping:
-	assert.Equal(t, "database.schema.foo", NewTableIdentifier("database", "schema", "foo", false).FullyQualifiedName())
-	assert.Equal(t, "database.schema.foo", NewTableIdentifier("database", "schema", "foo", true).FullyQualifiedName())
+	// Table name that is not a reserved word:
+	assert.Equal(t, "database.schema.foo", NewTableIdentifier("database", "schema", "foo").FullyQualifiedName())
 
-	// Table name that needs escaping:
-	assert.Equal(t, `database.schema."table"`, NewTableIdentifier("database", "schema", "table", false).FullyQualifiedName())
-	assert.Equal(t, `database.schema."TABLE"`, NewTableIdentifier("database", "schema", "table", true).FullyQualifiedName())
+	// Table name that is a reserved word:
+	assert.Equal(t, `database.schema."TABLE"`, NewTableIdentifier("database", "schema", "table").FullyQualifiedName())
 }
diff --git a/clients/snowflake/util_test.go b/clients/snowflake/util_test.go
index 63f3b4ad2..c7362f1ae 100644
--- a/clients/snowflake/util_test.go
+++ b/clients/snowflake/util_test.go
@@ -21,22 +21,22 @@ func TestAddPrefixToTableName(t *testing.T) {
 	testCases := []_testCase{
 		{
 			name:                "happy path",
-			tableID:             NewTableIdentifier("database", "schema", "tableName", true),
+			tableID:             NewTableIdentifier("database", "schema", "tableName"),
 			expectedFqTableName: "database.schema.%tableName",
 		},
 		{
 			name:                "tableName only",
-			tableID:             NewTableIdentifier("", "", "orders", true),
+			tableID:             NewTableIdentifier("", "", "orders"),
 			expectedFqTableName: "..%orders",
 		},
 		{
 			name:                "schema and tableName only",
-			tableID:             NewTableIdentifier("", "public", "orders", true),
+			tableID:             NewTableIdentifier("", "public", "orders"),
 			expectedFqTableName: ".public.%orders",
 		},
 		{
 			name:                "db and tableName only",
-			tableID:             NewTableIdentifier("db", "", "tableName", true),
+			tableID:             NewTableIdentifier("db", "", "tableName"),
 			expectedFqTableName: "db..%tableName",
 		},
 	}
diff --git a/lib/destination/ddl/ddl_create_table_test.go b/lib/destination/ddl/ddl_create_table_test.go
index a9ec46fcd..340649fd3 100644
--- a/lib/destination/ddl/ddl_create_table_test.go
+++ b/lib/destination/ddl/ddl_create_table_test.go
@@ -25,7 +25,7 @@ func (d *DDLTestSuite) Test_CreateTable() {
 	bqTableID := bigquery.NewTableIdentifier("", "mock_dataset", "mock_table")
 	d.bigQueryStore.GetConfigMap().AddTableToConfig(bqTableID, types.NewDwhTableConfig(&columns.Columns{}, nil, true, true))
 
-	snowflakeTableID := snowflake.NewTableIdentifier("", "mock_dataset", "mock_table", true)
+	snowflakeTableID := snowflake.NewTableIdentifier("", "mock_dataset", "mock_table")
 	d.snowflakeStagesStore.GetConfigMap().AddTableToConfig(snowflakeTableID, types.NewDwhTableConfig(&columns.Columns{}, nil, true, true))
 
 	type dwhToTableConfig struct {
@@ -114,7 +114,7 @@ func (d *DDLTestSuite) TestCreateTable() {
 	}
 
 	for index, testCase := range testCases {
-		tableID := snowflake.NewTableIdentifier("demo", "public", "experiments", false)
+		tableID := snowflake.NewTableIdentifier("demo", "public", "experiments")
 		d.snowflakeStagesStore.GetConfigMap().AddTableToConfig(tableID, types.NewDwhTableConfig(&columns.Columns{}, nil, true, true))
 		tc := d.snowflakeStagesStore.GetConfigMap().TableConfig(tableID)
 
diff --git a/lib/destination/ddl/ddl_sflk_test.go b/lib/destination/ddl/ddl_sflk_test.go
index 9921ccc40..f3e464cc9 100644
--- a/lib/destination/ddl/ddl_sflk_test.go
+++ b/lib/destination/ddl/ddl_sflk_test.go
@@ -29,7 +29,7 @@ func (d *DDLTestSuite) TestAlterComplexObjects() {
 		columns.NewColumn("select", typing.String),
 	}
 
-	tableID := snowflake.NewTableIdentifier("shop", "public", "complex_columns", true)
+	tableID := snowflake.NewTableIdentifier("shop", "public", "complex_columns")
 	fqTable := "shop.public.complex_columns"
 	d.snowflakeStagesStore.GetConfigMap().AddTableToConfig(tableID, types.NewDwhTableConfig(&columns.Columns{}, nil, false, true))
 	tc := d.snowflakeStagesStore.GetConfigMap().TableConfig(tableID)
@@ -64,7 +64,7 @@ func (d *DDLTestSuite) TestAlterIdempotency() {
 		columns.NewColumn("start", typing.String),
 	}
 
-	tableID := snowflake.NewTableIdentifier("shop", "public", "orders", true)
+	tableID := snowflake.NewTableIdentifier("shop", "public", "orders")
 	d.snowflakeStagesStore.GetConfigMap().AddTableToConfig(tableID, types.NewDwhTableConfig(&columns.Columns{}, nil, false, true))
 	tc := d.snowflakeStagesStore.GetConfigMap().TableConfig(tableID)
 
@@ -95,7 +95,7 @@ func (d *DDLTestSuite) TestAlterTableAdd() {
 		columns.NewColumn("start", typing.String),
 	}
 
-	tableID := snowflake.NewTableIdentifier("shop", "public", "orders", true)
+	tableID := snowflake.NewTableIdentifier("shop", "public", "orders")
 	d.snowflakeStagesStore.GetConfigMap().AddTableToConfig(tableID, types.NewDwhTableConfig(&columns.Columns{}, nil, false, true))
 	tc := d.snowflakeStagesStore.GetConfigMap().TableConfig(tableID)
 
@@ -138,7 +138,7 @@ func (d *DDLTestSuite) TestAlterTableDeleteDryRun() {
 		columns.NewColumn("start", typing.String),
 	}
 
-	tableID := snowflake.NewTableIdentifier("shop", "public", "users", true)
+	tableID := snowflake.NewTableIdentifier("shop", "public", "users")
 	fqTable := "shop.public.users"
 	d.snowflakeStagesStore.GetConfigMap().AddTableToConfig(tableID, types.NewDwhTableConfig(&columns.Columns{}, nil, false, true))
 	tc := d.snowflakeStagesStore.GetConfigMap().TableConfig(tableID)
@@ -198,7 +198,7 @@ func (d *DDLTestSuite) TestAlterTableDelete() {
 		columns.NewColumn("start", typing.String),
 	}
 
-	tableID := snowflake.NewTableIdentifier("shop", "public", "users1", true)
+	tableID := snowflake.NewTableIdentifier("shop", "public", "users1")
 
 	d.snowflakeStagesStore.GetConfigMap().AddTableToConfig(tableID, types.NewDwhTableConfig(&columns.Columns{}, map[string]time.Time{
 		"col_to_delete": time.Now().Add(-2 * constants.DeletionConfidencePadding),
diff --git a/lib/destination/ddl/ddl_temp_test.go b/lib/destination/ddl/ddl_temp_test.go
index 62d5ab0be..efef1bb96 100644
--- a/lib/destination/ddl/ddl_temp_test.go
+++ b/lib/destination/ddl/ddl_temp_test.go
@@ -35,7 +35,7 @@ func (d *DDLTestSuite) TestValidate_AlterTableArgs() {
 }
 
 func (d *DDLTestSuite) TestCreateTemporaryTable_Errors() {
-	tableID := snowflake.NewTableIdentifier("", "mock_dataset", "mock_table", false)
+	tableID := snowflake.NewTableIdentifier("", "mock_dataset", "mock_table")
 	d.snowflakeStagesStore.GetConfigMap().AddTableToConfig(tableID, types.NewDwhTableConfig(&columns.Columns{}, nil, true, true))
 	snowflakeTc := d.snowflakeStagesStore.GetConfigMap().TableConfig(tableID)
 	args := ddl.AlterTableArgs{
@@ -69,7 +69,7 @@ func (d *DDLTestSuite) TestCreateTemporaryTable_Errors() {
 func (d *DDLTestSuite) TestCreateTemporaryTable() {
 	{
 		// Snowflake Stage
-		tableID := snowflake.NewTableIdentifier("db", "schema", "tempTableName", false)
+		tableID := snowflake.NewTableIdentifier("db", "schema", "tempTableName")
 
 		d.snowflakeStagesStore.GetConfigMap().AddTableToConfig(tableID, types.NewDwhTableConfig(&columns.Columns{}, nil, true, true))
 		sflkStageTc := d.snowflakeStagesStore.GetConfigMap().TableConfig(tableID)