Merge branch 'master' into nv/rename-column-name

clients/snowflake/snowflake.go

@@ -128,10 +128,7 @@ func (s *Store) reestablishConnection() error {
 }
 
 func (s *Store) generateDedupeQueries(tableID, stagingTableID types.TableIdentifier, primaryKeys []string, topicConfig kafkalib.TopicConfig) []string {
-	var primaryKeysEscaped []string
-	for _, pk := range primaryKeys {
-		primaryKeysEscaped = append(primaryKeysEscaped, s.Dialect().QuoteIdentifier(pk))
-	}
+	primaryKeysEscaped := sql.QuoteIdentifiers(primaryKeys, s.Dialect())
 
 	orderColsToIterate := primaryKeysEscaped
 	if topicConfig.IncludeArtieUpdatedAt {

lib/sql/util.go

@@ -13,3 +13,11 @@ import (
 func QuoteLiteral(value string) string {
 	return fmt.Sprintf("'%s'", strings.ReplaceAll(stringutil.EscapeBackslashes(value), "'", `\'`))
 }
+
+func QuoteIdentifiers(identifiers []string, dialect Dialect) []string {
+	result := make([]string, len(identifiers))
+	for i, identifier := range identifiers {
+		result[i] = dialect.QuoteIdentifier(identifier)
+	}
+	return result
+}

lib/sql/util_test.go

@@ -38,3 +38,8 @@ func TestQuoteLiteral(t *testing.T) {
 		assert.Equal(t, testCase.expected, QuoteLiteral(testCase.colVal), testCase.name)
 	}
 }
+
+func TestQuoteIdentifiers(t *testing.T) {
+	assert.Equal(t, []string{}, QuoteIdentifiers([]string{}, BigQueryDialect{}))
+	assert.Equal(t, []string{"`a`", "`b`", "`c`"}, QuoteIdentifiers([]string{"a", "b", "c"}, BigQueryDialect{}))
+}