Clean up.

clients/snowflake/writes.go

@@ -4,6 +4,8 @@ import (
 	"fmt"
 	"log/slog"
 
+	"github.com/artie-labs/transfer/lib/sql"
+
 	"github.com/artie-labs/transfer/clients/shared"
 	"github.com/artie-labs/transfer/lib/destination/types"
 	"github.com/artie-labs/transfer/lib/logger"
@@ -52,9 +54,10 @@ func (s *Store) Merge(tableData *optimization.TableData) error {
 		}
 
 		var additionalEqualityStrings []string
-		if len(tableData.TopicConfig.AdditionalMergeColumns) > 0 {
-			for _, additionalMergeCol := range tableData.TopicConfig.AdditionalMergeColumns {
-				additionalEqualityStrings = append(additionalEqualityStrings, fmt.Sprintf("c.%s = cc.%s", additionalMergeCol, additionalMergeCol))
+		if len(tableData.TopicConfig().AdditionalMergePredicates) > 0 {
+			for _, additionalMergePredicate := range tableData.TopicConfig().AdditionalMergePredicates {
+				mergePredicateColumn := sql.EscapeName(additionalMergePredicate.PartitionField, s.ShouldUppercaseEscapedNames(), s.Label())
+				additionalEqualityStrings = append(additionalEqualityStrings, fmt.Sprintf("c.%s = cc.%s", mergePredicateColumn, mergePredicateColumn))
 			}
 		}
 

lib/kafkalib/partition/settings.go

@@ -17,6 +17,10 @@ var ValidPartitionBy = []string{
 	"daily",
 }
 
+type MergePredicates struct {
+	PartitionField string
+}
+
 type BigQuerySettings struct {
 	PartitionType  string `yaml:"partitionType" json:"partitionType"`
 	PartitionField string `yaml:"partitionField" json:"partitionField"`

lib/kafkalib/topic.go

@@ -47,7 +47,7 @@ type TopicConfig struct {
 	IncludeArtieUpdatedAt     bool                        `yaml:"includeArtieUpdatedAt"`
 	IncludeDatabaseUpdatedAt  bool                        `yaml:"includeDatabaseUpdatedAt"`
 	BigQueryPartitionSettings *partition.BigQuerySettings `yaml:"bigQueryPartitionSettings,omitempty"`
-	AdditionalMergeColumns    []string                    `yaml:"additionalMergeColumns,omitempty"`
+	AdditionalMergePredicates []partition.MergePredicates `yaml:"additionalMergePredicates,omitempty"`
 
 	// Internal metadata
 	opsToSkipMap map[string]bool `yaml:"-"`