Resync for SF and BQ: Quote column names for RenameTables (#1580)

This PR quotes column names for the insert into select command in
RenameTables activity for Snowflake and BigQuery. A column with name
'rows' would make the query fail due to it being a reserved keyword.
This PR fixes that

Functionally tested

flow/connectors/bigquery/bigquery.go

@@ -728,7 +728,7 @@ func (c *BigQueryConnector) RenameTables(ctx context.Context, req *protos.Rename
 
 		columnNames := make([]string, 0, len(renameRequest.TableSchema.Columns))
 		for _, col := range renameRequest.TableSchema.Columns {
-			columnNames = append(columnNames, col.Name)
+			columnNames = append(columnNames, "`"+col.Name+"`")
 		}
 
 		if req.SoftDeleteColName != nil {
@@ -744,7 +744,10 @@ func (c *BigQueryConnector) RenameTables(ctx context.Context, req *protos.Rename
 			allColsWithoutAlias := strings.Join(columnNames, ",")
 			allColsWithAlias := allColsBuilder.String()
 
-			pkeyCols := renameRequest.TableSchema.PrimaryKeyColumns
+			pkeyCols := make([]string, 0, len(renameRequest.TableSchema.PrimaryKeyColumns))
+			for _, pkeyCol := range renameRequest.TableSchema.PrimaryKeyColumns {
+				pkeyCols = append(pkeyCols, "`"+pkeyCol+"`")
+			}
 
 			c.logger.Info(fmt.Sprintf("handling soft-deletes for table '%s'...", dstDatasetTable.string()))
 

flow/connectors/snowflake/snowflake.go

@@ -755,11 +755,16 @@ func (c *SnowflakeConnector) RenameTables(ctx context.Context, req *protos.Renam
 
 			columnNames := make([]string, 0, len(renameRequest.TableSchema.Columns))
 			for _, col := range renameRequest.TableSchema.Columns {
-				columnNames = append(columnNames, col.Name)
+				columnNames = append(columnNames, SnowflakeIdentifierNormalize(col.Name))
+			}
+
+			pkeyColumnNames := make([]string, 0, len(renameRequest.TableSchema.PrimaryKeyColumns))
+			for _, col := range renameRequest.TableSchema.PrimaryKeyColumns {
+				pkeyColumnNames = append(pkeyColumnNames, SnowflakeIdentifierNormalize(col))
 			}
 
 			allCols := strings.Join(columnNames, ",")
-			pkeyCols := strings.Join(renameRequest.TableSchema.PrimaryKeyColumns, ",")
+			pkeyCols := strings.Join(pkeyColumnNames, ",")
 
 			c.logger.Info(fmt.Sprintf("handling soft-deletes for table '%s'...", dst))