Avoid double escape

flow/connectors/postgres/qrep.go

@@ -123,7 +123,7 @@ func (c *PostgresConnector) getNumRowsPartitions(
 
 	if totalRows.Int64 == 0 {
 		c.logger.Warn("no records to replicate, returning")
-		return make([]*protos.QRepPartition, 0), nil
+		return nil, nil
 	}
 
 	// Calculate the number of partitions
@@ -543,14 +543,15 @@ func syncQRepRecords(
 
 		// construct the SET clause for the upsert operation
 		upsertMatchColsList := writeMode.UpsertKeyColumns
-		upsertMatchCols := make(map[string]struct{})
+		upsertMatchCols := make(map[string]struct{}, len(upsertMatchColsList))
 		for _, col := range upsertMatchColsList {
 			upsertMatchCols[col] = struct{}{}
 		}
 
-		setClauseArray := make([]string, 0)
-		selectStrArray := make([]string, 0)
-		for _, col := range sink.GetColumnNames() {
+		columnNames := sink.GetColumnNames()
+		setClauseArray := make([]string, 0, len(upsertMatchColsList)+1)
+		selectStrArray := make([]string, 0, len(columnNames))
+		for _, col := range columnNames {
 			_, ok := upsertMatchCols[col]
 			quotedCol := QuoteIdentifier(col)
 			if !ok {

flow/connectors/postgres/sink_pg.go

@@ -72,7 +72,7 @@ func (p PgCopyWriter) ExecuteQueryWithTx(
 	fieldDescriptions := norows.FieldDescriptions()
 	cols := make([]string, 0, len(fieldDescriptions))
 	for _, fd := range fieldDescriptions {
-		cols = append(cols, QuoteIdentifier(fd.Name))
+		cols = append(cols, fd.Name)
 	}
 	p.SetSchema(cols)
 	norows.Close()
@@ -112,15 +112,20 @@ func (p PgCopyWriter) Close(err error) {
 }
 
 func (p PgCopyReader) GetColumnNames() []string {
+	<-p.schema.schemaLatch
 	return p.schema.schema
 }
 
 func (p PgCopyReader) CopyInto(ctx context.Context, c *PostgresConnector, tx pgx.Tx, table pgx.Identifier) (int64, error) {
-	<-p.schema.schemaLatch
+	cols := p.GetColumnNames()
+	quotedCols := make([]string, 0, len(cols))
+	for _, col := range cols {
+		quotedCols = append(cols, QuoteIdentifier(col))
+	}
 	ct, err := c.conn.PgConn().CopyFrom(
 		ctx,
 		p.PipeReader,
-		fmt.Sprintf("COPY %s (%s) FROM STDIN", table.Sanitize(), strings.Join(p.schema.schema, ",")),
+		fmt.Sprintf("COPY %s (%s) FROM STDIN", table.Sanitize(), strings.Join(quotedCols, ",")),
 	)
 	return ct.RowsAffected(), err
 }