Merge branch 'master' into nv/kill-column-rawname

clients/shared/merge.go

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

lib/destination/dml/merge.go

@@ -17,7 +17,7 @@ type MergeArgument struct {
 	TableID       types.TableIdentifier
 	SubQuery      string
 	IdempotentKey string
-	PrimaryKeys   []columns.Wrapper
+	PrimaryKeys   []columns.Column
 
 	// AdditionalEqualityStrings is used for handling BigQuery partitioned table merges
 	AdditionalEqualityStrings []string
@@ -94,7 +94,7 @@ func (m *MergeArgument) GetParts() ([]string, error) {
 	var equalitySQLParts []string
 	for _, primaryKey := range m.PrimaryKeys {
 		// We'll need to escape the primary key as well.
-		equalitySQL := fmt.Sprintf("c.%s = cc.%s", primaryKey.EscapedName(), primaryKey.EscapedName())
+		equalitySQL := fmt.Sprintf("c.%s = cc.%s", primaryKey.Name(m.Dialect), primaryKey.Name(m.Dialect))
 		equalitySQLParts = append(equalitySQLParts, equalitySQL)
 	}
 
@@ -115,7 +115,7 @@ func (m *MergeArgument) GetParts() ([]string, error) {
 				// LEFT JOIN table on pk(s)
 				m.TableID.FullyQualifiedName(), strings.Join(equalitySQLParts, " and "),
 				// Where PK is NULL (we only need to specify one primary key since it's covered with equalitySQL parts)
-				m.PrimaryKeys[0].EscapedName()),
+				m.PrimaryKeys[0].Name(m.Dialect)),
 			// UPDATE
 			fmt.Sprintf(`UPDATE %s as c SET %s FROM %s as cc WHERE %s%s;`,
 				// UPDATE table set col1 = cc. col1
@@ -142,7 +142,7 @@ func (m *MergeArgument) GetParts() ([]string, error) {
 
 	var pks []string
 	for _, pk := range m.PrimaryKeys {
-		pks = append(pks, pk.EscapedName())
+		pks = append(pks, pk.Name(m.Dialect))
 	}
 
 	parts := []string{
@@ -159,7 +159,7 @@ func (m *MergeArgument) GetParts() ([]string, error) {
 			// LEFT JOIN table on pk(s)
 			m.TableID.FullyQualifiedName(), strings.Join(equalitySQLParts, " and "),
 			// Where PK is NULL (we only need to specify one primary key since it's covered with equalitySQL parts)
-			m.PrimaryKeys[0].EscapedName()),
+			m.PrimaryKeys[0].Name(m.Dialect)),
 		// 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
@@ -207,15 +207,15 @@ func (m *MergeArgument) GetStatement() (string, error) {
 	var equalitySQLParts []string
 	for _, primaryKey := range m.PrimaryKeys {
 		// We'll need to escape the primary key as well.
-		equalitySQL := fmt.Sprintf("c.%s = cc.%s", primaryKey.EscapedName(), primaryKey.EscapedName())
+		equalitySQL := fmt.Sprintf("c.%s = cc.%s", primaryKey.Name(m.Dialect), primaryKey.Name(m.Dialect))
 		pkCol, isOk := m.Columns.GetColumn(primaryKey.RawName())
 		if !isOk {
 			return "", fmt.Errorf("column: %s does not exist in columnToType: %v", primaryKey.RawName(), m.Columns)
 		}
 
 		if m.DestKind == constants.BigQuery && pkCol.KindDetails.Kind == typing.Struct.Kind {
 			// BigQuery requires special casting to compare two JSON objects.
-			equalitySQL = fmt.Sprintf("TO_JSON_STRING(c.%s) = TO_JSON_STRING(cc.%s)", primaryKey.EscapedName(), primaryKey.EscapedName())
+			equalitySQL = fmt.Sprintf("TO_JSON_STRING(c.%s) = TO_JSON_STRING(cc.%s)", primaryKey.Name(m.Dialect), primaryKey.Name(m.Dialect))
 		}
 
 		equalitySQLParts = append(equalitySQLParts, equalitySQL)
@@ -295,7 +295,7 @@ func (m *MergeArgument) GetMSSQLStatement() (string, error) {
 	var equalitySQLParts []string
 	for _, primaryKey := range m.PrimaryKeys {
 		// We'll need to escape the primary key as well.
-		equalitySQL := fmt.Sprintf("c.%s = cc.%s", primaryKey.EscapedName(), primaryKey.EscapedName())
+		equalitySQL := fmt.Sprintf("c.%s = cc.%s", primaryKey.Name(m.Dialect), primaryKey.Name(m.Dialect))
 		equalitySQLParts = append(equalitySQLParts, equalitySQL)
 	}
 

lib/destination/dml/merge_bigquery_test.go

@@ -19,7 +19,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), sql.BigQueryDialect{})},
+		PrimaryKeys: []columns.Column{columns.NewColumn("order_id", typing.Invalid)},
 		Columns:     &cols,
 		DestKind:    constants.BigQuery,
 		Dialect:     sql.BigQueryDialect{},
@@ -41,7 +41,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), sql.BigQueryDialect{})},
+		PrimaryKeys: []columns.Column{columns.NewColumn("order_oid", typing.Invalid)},
 		Columns:     &cols,
 		DestKind:    constants.BigQuery,
 		Dialect:     sql.BigQueryDialect{},

lib/destination/dml/merge_mssql_test.go

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

lib/destination/dml/merge_parts_test.go

@@ -25,7 +25,7 @@ func TestMergeStatementPartsValidation(t *testing.T) {
 }
 
 type result struct {
-	PrimaryKeys    []columns.Wrapper
+	PrimaryKeys    []columns.Column
 	ColumnsToTypes columns.Columns
 }
 
@@ -47,11 +47,11 @@ func getBasicColumnsForTest(compositeKey bool) result {
 	cols.AddColumn(textToastCol)
 	cols.AddColumn(columns.NewColumn(constants.DeleteColumnMarker, typing.Boolean))
 
-	var pks []columns.Wrapper
-	pks = append(pks, columns.NewWrapper(idCol, sql.MSSQLDialect{}))
+	var pks []columns.Column
+	pks = append(pks, idCol)
 
 	if compositeKey {
-		pks = append(pks, columns.NewWrapper(emailCol, sql.MSSQLDialect{}))
+		pks = append(pks, emailCol)
 	}
 
 	return result{

lib/destination/dml/merge_test.go

@@ -56,16 +56,15 @@ func TestMergeStatementSoftDelete(t *testing.T) {
 	_cols.AddColumn(columns.NewColumn("id", typing.String))
 	_cols.AddColumn(columns.NewColumn(constants.DeleteColumnMarker, typing.Boolean))
 
-	dialect := sql.SnowflakeDialect{}
 	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), dialect)},
+			PrimaryKeys:   []columns.Column{columns.NewColumn("id", typing.Invalid)},
 			Columns:       &_cols,
 			DestKind:      constants.Snowflake,
-			Dialect:       dialect,
+			Dialect:       sql.SnowflakeDialect{},
 			SoftDelete:    true,
 		}
 
@@ -107,15 +106,14 @@ 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{}
 	mergeArg := MergeArgument{
 		TableID:       MockTableIdentifier{fqTable},
 		SubQuery:      subQuery,
 		IdempotentKey: "",
-		PrimaryKeys:   []columns.Wrapper{columns.NewWrapper(columns.NewColumn("id", typing.Invalid), dialect)},
+		PrimaryKeys:   []columns.Column{columns.NewColumn("id", typing.Invalid)},
 		Columns:       &_cols,
 		DestKind:      constants.Snowflake,
-		Dialect:       dialect,
+		Dialect:       sql.SnowflakeDialect{},
 		SoftDelete:    false,
 	}
 
@@ -156,15 +154,14 @@ func TestMergeStatementIdempotentKey(t *testing.T) {
 	_cols.AddColumn(columns.NewColumn("id", typing.String))
 	_cols.AddColumn(columns.NewColumn(constants.DeleteColumnMarker, typing.Boolean))
 
-	dialect := sql.SnowflakeDialect{}
 	mergeArg := MergeArgument{
 		TableID:       MockTableIdentifier{fqTable},
 		SubQuery:      subQuery,
 		IdempotentKey: "updated_at",
-		PrimaryKeys:   []columns.Wrapper{columns.NewWrapper(columns.NewColumn("id", typing.Invalid), dialect)},
+		PrimaryKeys:   []columns.Column{columns.NewColumn("id", typing.Invalid)},
 		Columns:       &_cols,
 		DestKind:      constants.Snowflake,
-		Dialect:       dialect,
+		Dialect:       sql.SnowflakeDialect{},
 		SoftDelete:    false,
 	}
 
@@ -199,18 +196,17 @@ func TestMergeStatementCompositeKey(t *testing.T) {
 	_cols.AddColumn(columns.NewColumn("another_id", typing.String))
 	_cols.AddColumn(columns.NewColumn(constants.DeleteColumnMarker, typing.Boolean))
 
-	dialect := sql.SnowflakeDialect{}
 	mergeArg := MergeArgument{
 		TableID:       MockTableIdentifier{fqTable},
 		SubQuery:      subQuery,
 		IdempotentKey: "updated_at",
-		PrimaryKeys: []columns.Wrapper{
-			columns.NewWrapper(columns.NewColumn("id", typing.Invalid), dialect),
-			columns.NewWrapper(columns.NewColumn("another_id", typing.Invalid), dialect),
+		PrimaryKeys: []columns.Column{
+			columns.NewColumn("id", typing.Invalid),
+			columns.NewColumn("another_id", typing.Invalid),
 		},
 		Columns:    &_cols,
 		DestKind:   constants.Snowflake,
-		Dialect:    dialect,
+		Dialect:    sql.SnowflakeDialect{},
 		SoftDelete: false,
 	}
 
@@ -249,18 +245,17 @@ 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{}
 	mergeArg := MergeArgument{
 		TableID:       MockTableIdentifier{fqTable},
 		SubQuery:      subQuery,
 		IdempotentKey: "",
-		PrimaryKeys: []columns.Wrapper{
-			columns.NewWrapper(columns.NewColumn("id", typing.Invalid), dialect),
-			columns.NewWrapper(columns.NewColumn("group", typing.Invalid), dialect),
+		PrimaryKeys: []columns.Column{
+			columns.NewColumn("id", typing.Invalid),
+			columns.NewColumn("group", typing.Invalid),
 		},
 		Columns:    &_cols,
 		DestKind:   constants.Snowflake,
-		Dialect:    dialect,
+		Dialect:    sql.SnowflakeDialect{},
 		SoftDelete: false,
 	}
 

lib/destination/dml/merge_valid_test.go

@@ -13,8 +13,8 @@ import (
 )
 
 func TestMergeArgument_Valid(t *testing.T) {
-	primaryKeys := []columns.Wrapper{
-		columns.NewWrapper(columns.NewColumn("id", typing.Integer), sql.SnowflakeDialect{}),
+	primaryKeys := []columns.Column{
+		columns.NewColumn("id", typing.Integer),
 	}
 
 	var cols columns.Columns

lib/optimization/table_data.go

@@ -10,7 +10,6 @@ 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"
@@ -67,10 +66,10 @@ func (t *TableData) ContainOtherOperations() bool {
 	return t.containOtherOperations
 }
 
-func (t *TableData) PrimaryKeys(dialect sql.Dialect) []columns.Wrapper {
-	var pks []columns.Wrapper
+func (t *TableData) PrimaryKeys() []columns.Column {
+	var pks []columns.Column
 	for _, pk := range t.primaryKeys {
-		pks = append(pks, columns.NewWrapper(columns.NewColumn(pk, typing.Invalid), dialect))
+		pks = append(pks, columns.NewColumn(pk, typing.Invalid))
 	}
 
 	return pks