error out if duplicates

flow/connectors/bigquery/bigquery.go

@@ -613,6 +613,12 @@ func (c *BigQueryConnector) NormalizeRecords(req *model.NormalizeRecordsRequest)
 			},
 			shortColumn: map[string]string{},
 		}
+
+		// if there are duplicates in the unchangedToastColumns, error out here.
+		if len(unchangedToastColumns) != len(utils.Unique(unchangedToastColumns)) {
+			return nil, fmt.Errorf("duplicates found in unchangedToastColumns for table %s", tableName)
+		}
+
 		// normalize anything between last normalized batch id to last sync batchid
 		mergeStmts := mergeGen.generateMergeStmts(unchangedToastColumns)
 		for i, mergeStmt := range mergeStmts {

flow/connectors/utils/array.go

@@ -17,6 +17,18 @@ func ArrayMinus(first []string, second []string) []string {
 	return result
 }
 
+func Unique[T comparable](slice []T) []T {
+	keys := make(map[T]bool)
+	list := []T{}
+	for _, entry := range slice {
+		if _, value := keys[entry]; !value {
+			keys[entry] = true
+			list = append(list, entry)
+		}
+	}
+	return list
+}
+
 func ArrayChunks[T any](slice []T, size int) [][]T {
 	var partitions [][]T