Pass Dialect down (#529)

Signed-off-by: Nathan <[email protected]>

clients/shared/merge.go

@@ -125,7 +125,7 @@ func Merge(dwh destination.DataWarehouse, tableData *optimization.TableData, cfg
 		TableID:             tableID,
 		SubQuery:            subQuery,
 		IdempotentKey:       tableData.TopicConfig().IdempotentKey,
-		PrimaryKeys:         tableData.PrimaryKeys(dwh.ShouldUppercaseEscapedNames(), dwh.Label()),
+		PrimaryKeys:         tableData.PrimaryKeys(dwh.Dialect()),
 		Columns:             tableData.ReadOnlyInMemoryCols(),
 		SoftDelete:          tableData.TopicConfig().SoftDelete,
 		DestKind:            dwh.Label(),

clients/shared/utils.go

@@ -30,7 +30,7 @@ func BackfillColumn(cfg config.Config, dwh destination.DataWarehouse, column col
 		return fmt.Errorf("failed to escape default value: %w", err)
 	}
 
-	escapedCol := column.Name(dwh.ShouldUppercaseEscapedNames(), dwh.Label())
+	escapedCol := column.Name(dwh.Dialect())
 
 	// TODO: This is added because `default` is not technically a column that requires escaping, but it is required when it's in the where clause.
 	// Once we escape everything by default, we can remove this patch of code.

clients/snowflake/staging.go

@@ -85,7 +85,7 @@ func (s *Store) PrepareTemporaryTable(tableData *optimization.TableData, tableCo
 	// COPY the CSV file (in Snowflake) into a table
 	copyCommand := fmt.Sprintf("COPY INTO %s (%s) FROM (SELECT %s FROM @%s)",
 		tempTableID.FullyQualifiedName(),
-		strings.Join(tableData.ReadOnlyInMemoryCols().GetEscapedColumnsToUpdate(s.ShouldUppercaseEscapedNames(), s.Label()), ","),
+		strings.Join(tableData.ReadOnlyInMemoryCols().GetEscapedColumnsToUpdate(s.Dialect()), ","),
 		escapeColumns(tableData.ReadOnlyInMemoryCols(), ","), addPrefixToTableName(tempTableID, "%"))
 
 	if additionalSettings.AdditionalCopyClause != "" {

lib/destination/ddl/ddl.go

@@ -104,7 +104,7 @@ func (a AlterTableArgs) AlterTable(cols ...columns.Column) error {
 		mutateCol = append(mutateCol, col)
 		switch a.ColumnOp {
 		case constants.Add:
-			colName := col.Name(*a.UppercaseEscNames, a.Dwh.Label())
+			colName := col.Name(a.Dwh.Dialect())
 
 			if col.PrimaryKey() && a.Mode != config.History {
 				// Don't create a PK for history mode because it's append-only, so the primary key should not be enforced.
@@ -113,7 +113,7 @@ func (a AlterTableArgs) AlterTable(cols ...columns.Column) error {
 
 			colSQLParts = append(colSQLParts, fmt.Sprintf(`%s %s`, colName, typing.KindToDWHType(col.KindDetails, a.Dwh.Label(), col.PrimaryKey())))
 		case constants.Delete:
-			colSQLParts = append(colSQLParts, col.Name(*a.UppercaseEscNames, a.Dwh.Label()))
+			colSQLParts = append(colSQLParts, col.Name(a.Dwh.Dialect()))
 		}
 	}
 

lib/destination/ddl/ddl_bq_test.go

@@ -90,7 +90,7 @@ func (d *DDLTestSuite) TestAlterTableDropColumnsBigQuery() {
 
 		assert.NoError(d.T(), alterTableArgs.AlterTable(column))
 		query, _ := d.fakeBigQueryStore.ExecArgsForCall(callIdx)
-		assert.Equal(d.T(), fmt.Sprintf("ALTER TABLE %s drop COLUMN %s", fqName, column.Name(false, d.bigQueryStore.Label())), query)
+		assert.Equal(d.T(), fmt.Sprintf("ALTER TABLE %s drop COLUMN %s", fqName, column.Name(d.bigQueryStore.Dialect())), query)
 		callIdx += 1
 	}
 
@@ -148,7 +148,7 @@ func (d *DDLTestSuite) TestAlterTableAddColumns() {
 
 		assert.NoError(d.T(), alterTableArgs.AlterTable(col))
 		query, _ := d.fakeBigQueryStore.ExecArgsForCall(callIdx)
-		assert.Equal(d.T(), fmt.Sprintf("ALTER TABLE %s %s COLUMN %s %s", fqName, constants.Add, col.Name(false, d.bigQueryStore.Label()),
+		assert.Equal(d.T(), fmt.Sprintf("ALTER TABLE %s %s COLUMN %s %s", fqName, constants.Add, col.Name(d.bigQueryStore.Dialect()),
 			typing.KindToDWHType(kind, d.bigQueryStore.Label(), false)), query)
 		callIdx += 1
 	}
@@ -208,7 +208,7 @@ func (d *DDLTestSuite) TestAlterTableAddColumnsSomeAlreadyExist() {
 
 		assert.NoError(d.T(), alterTableArgs.AlterTable(column))
 		query, _ := d.fakeBigQueryStore.ExecArgsForCall(callIdx)
-		assert.Equal(d.T(), fmt.Sprintf("ALTER TABLE %s %s COLUMN %s %s", fqName, constants.Add, column.Name(false, d.bigQueryStore.Label()),
+		assert.Equal(d.T(), fmt.Sprintf("ALTER TABLE %s %s COLUMN %s %s", fqName, constants.Add, column.Name(d.bigQueryStore.Dialect()),
 			typing.KindToDWHType(column.KindDetails, d.bigQueryStore.Label(), false)), query)
 		callIdx += 1
 	}

lib/destination/ddl/ddl_sflk_test.go

@@ -47,7 +47,7 @@ func (d *DDLTestSuite) TestAlterComplexObjects() {
 	for i := 0; i < len(cols); i++ {
 		execQuery, _ := d.fakeSnowflakeStagesStore.ExecArgsForCall(i)
 		assert.Equal(d.T(), fmt.Sprintf("ALTER TABLE %s add COLUMN %s %s", `shop.public."COMPLEX_COLUMNS"`,
-			cols[i].Name(false, d.snowflakeStagesStore.Label()),
+			cols[i].Name(d.snowflakeStagesStore.Dialect()),
 			typing.KindToDWHType(cols[i].KindDetails, d.snowflakeStagesStore.Label(), false)), execQuery)
 	}
 
@@ -180,7 +180,7 @@ func (d *DDLTestSuite) TestAlterTableDeleteDryRun() {
 
 		execArg, _ := d.fakeSnowflakeStagesStore.ExecArgsForCall(i)
 		assert.Equal(d.T(), execArg, fmt.Sprintf("ALTER TABLE %s %s COLUMN %s", `shop.public."USERS"`, constants.Delete,
-			cols[i].Name(false, d.snowflakeStagesStore.Label())))
+			cols[i].Name(d.snowflakeStagesStore.Dialect())))
 	}
 }
 

lib/destination/dml/merge.go

@@ -103,7 +103,7 @@ func (m *MergeArgument) GetParts() ([]string, error) {
 		equalitySQLParts = append(equalitySQLParts, equalitySQL)
 	}
 
-	cols := m.Columns.GetEscapedColumnsToUpdate(*m.UppercaseEscNames, m.DestKind)
+	cols := m.Columns.GetEscapedColumnsToUpdate(m.Dialect)
 
 	if m.SoftDelete {
 		return []string{
@@ -124,7 +124,7 @@ func (m *MergeArgument) GetParts() ([]string, error) {
 			// UPDATE
 			fmt.Sprintf(`UPDATE %s as c SET %s FROM %s as cc WHERE %s%s;`,
 				// UPDATE table set col1 = cc. col1
-				m.TableID.FullyQualifiedName(), m.Columns.UpdateQuery(m.DestKind, *m.UppercaseEscNames, false),
+				m.TableID.FullyQualifiedName(), m.Columns.UpdateQuery(m.DestKind, m.Dialect, false),
 				// FROM table (temp) WHERE join on PK(s)
 				m.SubQuery, strings.Join(equalitySQLParts, " and "), idempotentClause,
 			),
@@ -168,7 +168,7 @@ func (m *MergeArgument) GetParts() ([]string, error) {
 		// UPDATE
 		fmt.Sprintf(`UPDATE %s as c SET %s FROM %s as cc WHERE %s%s AND COALESCE(cc.%s, false) = false;`,
 			// UPDATE table set col1 = cc. col1
-			m.TableID.FullyQualifiedName(), m.Columns.UpdateQuery(m.DestKind, *m.UppercaseEscNames, true),
+			m.TableID.FullyQualifiedName(), m.Columns.UpdateQuery(m.DestKind, m.Dialect, true),
 			// FROM staging WHERE join on PK(s)
 			m.SubQuery, strings.Join(equalitySQLParts, " and "), idempotentClause, constants.DeleteColumnMarker,
 		),
@@ -235,7 +235,7 @@ func (m *MergeArgument) GetStatement() (string, error) {
 		equalitySQLParts = append(equalitySQLParts, m.AdditionalEqualityStrings...)
 	}
 
-	cols := m.Columns.GetEscapedColumnsToUpdate(*m.UppercaseEscNames, m.DestKind)
+	cols := m.Columns.GetEscapedColumnsToUpdate(m.Dialect)
 
 	if m.SoftDelete {
 		return fmt.Sprintf(`
@@ -244,7 +244,7 @@ WHEN MATCHED %sTHEN UPDATE SET %s
 WHEN NOT MATCHED AND IFNULL(cc.%s, false) = false THEN INSERT (%s) VALUES (%s);`,
 			m.TableID.FullyQualifiedName(), subQuery, strings.Join(equalitySQLParts, " and "),
 			// Update + Soft Deletion
-			idempotentClause, m.Columns.UpdateQuery(m.DestKind, *m.UppercaseEscNames, false),
+			idempotentClause, m.Columns.UpdateQuery(m.DestKind, m.Dialect, false),
 			// Insert
 			constants.DeleteColumnMarker, strings.Join(cols, ","),
 			array.StringsJoinAddPrefix(array.StringsJoinAddPrefixArgs{
@@ -277,7 +277,7 @@ WHEN NOT MATCHED AND IFNULL(cc.%s, false) = false THEN INSERT (%s) VALUES (%s);`
 		// Delete
 		constants.DeleteColumnMarker,
 		// Update
-		constants.DeleteColumnMarker, idempotentClause, m.Columns.UpdateQuery(m.DestKind, *m.UppercaseEscNames, true),
+		constants.DeleteColumnMarker, idempotentClause, m.Columns.UpdateQuery(m.DestKind, m.Dialect, true),
 		// Insert
 		constants.DeleteColumnMarker, strings.Join(cols, ","),
 		array.StringsJoinAddPrefix(array.StringsJoinAddPrefixArgs{
@@ -304,7 +304,7 @@ func (m *MergeArgument) GetMSSQLStatement() (string, error) {
 		equalitySQLParts = append(equalitySQLParts, equalitySQL)
 	}
 
-	cols := m.Columns.GetEscapedColumnsToUpdate(*m.UppercaseEscNames, m.DestKind)
+	cols := m.Columns.GetEscapedColumnsToUpdate(m.Dialect)
 
 	if m.SoftDelete {
 		return fmt.Sprintf(`
@@ -314,7 +314,7 @@ WHEN MATCHED %sTHEN UPDATE SET %s
 WHEN NOT MATCHED AND COALESCE(cc.%s, 0) = 0 THEN INSERT (%s) VALUES (%s);`,
 			m.TableID.FullyQualifiedName(), m.SubQuery, strings.Join(equalitySQLParts, " and "),
 			// Update + Soft Deletion
-			idempotentClause, m.Columns.UpdateQuery(m.DestKind, *m.UppercaseEscNames, false),
+			idempotentClause, m.Columns.UpdateQuery(m.DestKind, m.Dialect, false),
 			// Insert
 			constants.DeleteColumnMarker, strings.Join(cols, ","),
 			array.StringsJoinAddPrefix(array.StringsJoinAddPrefixArgs{
@@ -348,7 +348,7 @@ WHEN NOT MATCHED AND COALESCE(cc.%s, 1) = 0 THEN INSERT (%s) VALUES (%s);`,
 		// Delete
 		constants.DeleteColumnMarker,
 		// Update
-		constants.DeleteColumnMarker, idempotentClause, m.Columns.UpdateQuery(m.DestKind, *m.UppercaseEscNames, true),
+		constants.DeleteColumnMarker, idempotentClause, m.Columns.UpdateQuery(m.DestKind, m.Dialect, true),
 		// Insert
 		constants.DeleteColumnMarker, strings.Join(cols, ","),
 		array.StringsJoinAddPrefix(array.StringsJoinAddPrefixArgs{

lib/destination/dml/merge_bigquery_test.go

@@ -20,7 +20,7 @@ func TestMergeStatement_TempTable(t *testing.T) {
 	mergeArg := &MergeArgument{
 		TableID:           MockTableIdentifier{"customers.orders"},
 		SubQuery:          "customers.orders_tmp",
-		PrimaryKeys:       []columns.Wrapper{columns.NewWrapper(columns.NewColumn("order_id", typing.Invalid), false, constants.BigQuery)},
+		PrimaryKeys:       []columns.Wrapper{columns.NewWrapper(columns.NewColumn("order_id", typing.Invalid), sql.BigQueryDialect{})},
 		Columns:           &cols,
 		DestKind:          constants.BigQuery,
 		Dialect:           sql.BigQueryDialect{},
@@ -43,7 +43,7 @@ func TestMergeStatement_JSONKey(t *testing.T) {
 	mergeArg := &MergeArgument{
 		TableID:           MockTableIdentifier{"customers.orders"},
 		SubQuery:          "customers.orders_tmp",
-		PrimaryKeys:       []columns.Wrapper{columns.NewWrapper(columns.NewColumn("order_oid", typing.Invalid), false, constants.BigQuery)},
+		PrimaryKeys:       []columns.Wrapper{columns.NewWrapper(columns.NewColumn("order_oid", typing.Invalid), sql.BigQueryDialect{})},
 		Columns:           &cols,
 		DestKind:          constants.BigQuery,
 		Dialect:           sql.BigQueryDialect{},

lib/destination/dml/merge_mssql_test.go

@@ -45,7 +45,7 @@ func Test_GetMSSQLStatement(t *testing.T) {
 		TableID:           MockTableIdentifier{fqTable},
 		SubQuery:          subQuery,
 		IdempotentKey:     "",
-		PrimaryKeys:       []columns.Wrapper{columns.NewWrapper(columns.NewColumn("id", typing.Invalid), false, constants.MSSQL)},
+		PrimaryKeys:       []columns.Wrapper{columns.NewWrapper(columns.NewColumn("id", typing.Invalid), sql.DefaultDialect{})},
 		Columns:           &_cols,
 		DestKind:          constants.MSSQL,
 		Dialect:           sql.DefaultDialect{},

lib/destination/dml/merge_parts_test.go

@@ -48,10 +48,10 @@ func getBasicColumnsForTest(compositeKey bool) result {
 	cols.AddColumn(columns.NewColumn(constants.DeleteColumnMarker, typing.Boolean))
 
 	var pks []columns.Wrapper
-	pks = append(pks, columns.NewWrapper(idCol, false, constants.Redshift))
+	pks = append(pks, columns.NewWrapper(idCol, sql.DefaultDialect{}))
 
 	if compositeKey {
-		pks = append(pks, columns.NewWrapper(emailCol, false, constants.Redshift))
+		pks = append(pks, columns.NewWrapper(emailCol, sql.DefaultDialect{}))
 	}
 
 	return result{

lib/destination/dml/merge_test.go

@@ -57,15 +57,16 @@ func TestMergeStatementSoftDelete(t *testing.T) {
 	_cols.AddColumn(columns.NewColumn("id", typing.String))
 	_cols.AddColumn(columns.NewColumn(constants.DeleteColumnMarker, typing.Boolean))
 
+	dialect := sql.SnowflakeDialect{UppercaseEscNames: true}
 	for _, idempotentKey := range []string{"", "updated_at"} {
 		mergeArg := MergeArgument{
 			TableID:           MockTableIdentifier{fqTable},
 			SubQuery:          subQuery,
 			IdempotentKey:     idempotentKey,
-			PrimaryKeys:       []columns.Wrapper{columns.NewWrapper(columns.NewColumn("id", typing.Invalid), true, constants.Snowflake)},
+			PrimaryKeys:       []columns.Wrapper{columns.NewWrapper(columns.NewColumn("id", typing.Invalid), dialect)},
 			Columns:           &_cols,
 			DestKind:          constants.Snowflake,
-			Dialect:           sql.SnowflakeDialect{UppercaseEscNames: true},
+			Dialect:           dialect,
 			SoftDelete:        true,
 			UppercaseEscNames: ptr.ToBool(true),
 		}
@@ -108,14 +109,15 @@ func TestMergeStatement(t *testing.T) {
 	subQuery := fmt.Sprintf("SELECT %s from (values %s) as %s(%s)",
 		strings.Join(cols, ","), strings.Join(tableValues, ","), "_tbl", strings.Join(cols, ","))
 
+	dialect := sql.SnowflakeDialect{UppercaseEscNames: true}
 	mergeArg := MergeArgument{
 		TableID:           MockTableIdentifier{fqTable},
 		SubQuery:          subQuery,
 		IdempotentKey:     "",
-		PrimaryKeys:       []columns.Wrapper{columns.NewWrapper(columns.NewColumn("id", typing.Invalid), true, constants.Snowflake)},
+		PrimaryKeys:       []columns.Wrapper{columns.NewWrapper(columns.NewColumn("id", typing.Invalid), dialect)},
 		Columns:           &_cols,
 		DestKind:          constants.Snowflake,
-		Dialect:           sql.SnowflakeDialect{UppercaseEscNames: true},
+		Dialect:           dialect,
 		SoftDelete:        false,
 		UppercaseEscNames: ptr.ToBool(true),
 	}
@@ -157,14 +159,15 @@ func TestMergeStatementIdempotentKey(t *testing.T) {
 	_cols.AddColumn(columns.NewColumn("id", typing.String))
 	_cols.AddColumn(columns.NewColumn(constants.DeleteColumnMarker, typing.Boolean))
 
+	dialect := sql.SnowflakeDialect{UppercaseEscNames: true}
 	mergeArg := MergeArgument{
 		TableID:           MockTableIdentifier{fqTable},
 		SubQuery:          subQuery,
 		IdempotentKey:     "updated_at",
-		PrimaryKeys:       []columns.Wrapper{columns.NewWrapper(columns.NewColumn("id", typing.Invalid), true, constants.Snowflake)},
+		PrimaryKeys:       []columns.Wrapper{columns.NewWrapper(columns.NewColumn("id", typing.Invalid), dialect)},
 		Columns:           &_cols,
 		DestKind:          constants.Snowflake,
-		Dialect:           sql.SnowflakeDialect{UppercaseEscNames: true},
+		Dialect:           dialect,
 		SoftDelete:        false,
 		UppercaseEscNames: ptr.ToBool(true),
 	}
@@ -200,17 +203,18 @@ func TestMergeStatementCompositeKey(t *testing.T) {
 	_cols.AddColumn(columns.NewColumn("another_id", typing.String))
 	_cols.AddColumn(columns.NewColumn(constants.DeleteColumnMarker, typing.Boolean))
 
+	dialect := sql.SnowflakeDialect{UppercaseEscNames: true}
 	mergeArg := MergeArgument{
 		TableID:       MockTableIdentifier{fqTable},
 		SubQuery:      subQuery,
 		IdempotentKey: "updated_at",
 		PrimaryKeys: []columns.Wrapper{
-			columns.NewWrapper(columns.NewColumn("id", typing.Invalid), true, constants.Snowflake),
-			columns.NewWrapper(columns.NewColumn("another_id", typing.Invalid), true, constants.Snowflake),
+			columns.NewWrapper(columns.NewColumn("id", typing.Invalid), dialect),
+			columns.NewWrapper(columns.NewColumn("another_id", typing.Invalid), dialect),
 		},
 		Columns:           &_cols,
 		DestKind:          constants.Snowflake,
-		Dialect:           sql.SnowflakeDialect{UppercaseEscNames: true},
+		Dialect:           dialect,
 		SoftDelete:        false,
 		UppercaseEscNames: ptr.ToBool(true),
 	}
@@ -250,17 +254,18 @@ func TestMergeStatementEscapePrimaryKeys(t *testing.T) {
 	subQuery := fmt.Sprintf("SELECT %s from (values %s) as %s(%s)",
 		strings.Join(cols, ","), strings.Join(tableValues, ","), "_tbl", strings.Join(cols, ","))
 
+	dialect := sql.SnowflakeDialect{UppercaseEscNames: true}
 	mergeArg := MergeArgument{
 		TableID:       MockTableIdentifier{fqTable},
 		SubQuery:      subQuery,
 		IdempotentKey: "",
 		PrimaryKeys: []columns.Wrapper{
-			columns.NewWrapper(columns.NewColumn("id", typing.Invalid), true, constants.Snowflake),
-			columns.NewWrapper(columns.NewColumn("group", typing.Invalid), true, constants.Snowflake),
+			columns.NewWrapper(columns.NewColumn("id", typing.Invalid), dialect),
+			columns.NewWrapper(columns.NewColumn("group", typing.Invalid), dialect),
 		},
 		Columns:           &_cols,
 		DestKind:          constants.Snowflake,
-		Dialect:           sql.SnowflakeDialect{UppercaseEscNames: true},
+		Dialect:           dialect,
 		SoftDelete:        false,
 		UppercaseEscNames: ptr.ToBool(true),
 	}

lib/destination/dml/merge_valid_test.go

@@ -15,7 +15,7 @@ import (
 
 func TestMergeArgument_Valid(t *testing.T) {
 	primaryKeys := []columns.Wrapper{
-		columns.NewWrapper(columns.NewColumn("id", typing.Integer), true, constants.Snowflake),
+		columns.NewWrapper(columns.NewColumn("id", typing.Integer), sql.SnowflakeDialect{UppercaseEscNames: true}),
 	}
 
 	var cols columns.Columns

lib/optimization/table_data.go

@@ -10,6 +10,7 @@ import (
 	"github.com/artie-labs/transfer/lib/config/constants"
 	"github.com/artie-labs/transfer/lib/kafkalib"
 	"github.com/artie-labs/transfer/lib/size"
+	"github.com/artie-labs/transfer/lib/sql"
 	"github.com/artie-labs/transfer/lib/stringutil"
 	"github.com/artie-labs/transfer/lib/typing"
 	"github.com/artie-labs/transfer/lib/typing/columns"
@@ -66,10 +67,10 @@ func (t *TableData) ContainOtherOperations() bool {
 	return t.containOtherOperations
 }
 
-func (t *TableData) PrimaryKeys(uppercaseEscNames bool, destKind constants.DestinationKind) []columns.Wrapper {
+func (t *TableData) PrimaryKeys(dialect sql.Dialect) []columns.Wrapper {
 	var pks []columns.Wrapper
 	for _, pk := range t.primaryKeys {
-		pks = append(pks, columns.NewWrapper(columns.NewColumn(pk, typing.Invalid), uppercaseEscNames, destKind))
+		pks = append(pks, columns.NewWrapper(columns.NewColumn(pk, typing.Invalid), dialect))
 	}
 
 	return pks

lib/sql/escape.go

@@ -1,35 +1,8 @@
 package sql
 
-import (
-	"github.com/artie-labs/transfer/lib/config/constants"
-)
-
-func EscapeNameIfNecessary(name string, uppercaseEscNames bool, destKind constants.DestinationKind) string {
-	// TODO: Switch all calls of [EscapeNameIfNecessary] to [EscapeNameIfNecessaryUsingDialect] and kill this.
-	var dialect = dialectFor(destKind, uppercaseEscNames)
-
-	if destKind != constants.S3 && dialect.NeedsEscaping(name) {
-		return dialect.QuoteIdentifier(name)
-	}
-	return name
-}
-
 func EscapeNameIfNecessaryUsingDialect(name string, dialect Dialect) string {
 	if dialect.NeedsEscaping(name) {
 		return dialect.QuoteIdentifier(name)
 	}
 	return name
 }
-
-func dialectFor(destKind constants.DestinationKind, uppercaseEscNames bool) Dialect {
-	switch destKind {
-	case constants.BigQuery:
-		return BigQueryDialect{}
-	case constants.Snowflake:
-		return SnowflakeDialect{UppercaseEscNames: uppercaseEscNames}
-	case constants.Redshift:
-		return RedshiftDialect{}
-	default:
-		return DefaultDialect{}
-	}
-}