fix: Fix number of writes reported by postgresql (#6757)

hermanschaaf · web-flow · commit 018edf44322f · 2023-01-12T14:15:38.000Z
Because we didn't use the batch size, but rather the batch size limit when incrementing the counter for number of records written, the results may have been inaccurate in some cases.
diff --git a/plugins/destination/postgresql/client/write.go b/plugins/destination/postgresql/client/write.go
@@ -69,7 +69,8 @@ func (c *Client) Write(ctx context.Context, tables schema.Tables, res <-chan *de
 			sql = c.upsert(table)
 		}
 		batch.Queue(sql, r.Data...)
-		if batch.Len() >= c.batchSize {
+		batchSize := batch.Len()
+		if batchSize >= c.batchSize {
 			br := c.conn.SendBatch(ctx, batch)
 			if err := br.Close(); err != nil {
 				var pgErr *pgconn.PgError
@@ -79,12 +80,13 @@ func (c *Client) Write(ctx context.Context, tables schema.Tables, res <-chan *de
 				}
 				return fmt.Errorf("failed to execute batch with pgerror: %s: %w", pgErrToStr(pgErr), err)
 			}
-			atomic.AddUint64(&c.metrics.Writes, uint64(c.batchSize))
+			atomic.AddUint64(&c.metrics.Writes, uint64(batchSize))
 			batch = &pgx.Batch{}
 		}
 	}
 
-	if batch.Len() > 0 {
+	batchSize := batch.Len()
+	if batchSize > 0 {
 		br := c.conn.SendBatch(ctx, batch)
 		if err := br.Close(); err != nil {
 			var pgErr *pgconn.PgError
@@ -94,7 +96,7 @@ func (c *Client) Write(ctx context.Context, tables schema.Tables, res <-chan *de
 			}
 			return fmt.Errorf("failed to execute batch with pgerror: %s: %w", pgErrToStr(pgErr), err)
 		}
-		atomic.AddUint64(&c.metrics.Writes, uint64(c.batchSize))
+		atomic.AddUint64(&c.metrics.Writes, uint64(batchSize))
 	}
 
 	return nil

Original file line number	Diff line number	Diff line change
`@@ -69,7 +69,8 @@ func (c Client) Write(ctx context.Context, tables schema.Tables, res <-chan de`
`69`	`69`	`sql = c.upsert(table)`
`70`	`70`	`}`
`71`	`71`	`batch.Queue(sql, r.Data...)`
`72`		`- if batch.Len() >= c.batchSize {`
	`72`	`+ batchSize := batch.Len()`
	`73`	`+ if batchSize >= c.batchSize {`
`73`	`74`	`br := c.conn.SendBatch(ctx, batch)`
`74`	`75`	`if err := br.Close(); err != nil {`
`75`	`76`	`var pgErr *pgconn.PgError`
`@@ -79,12 +80,13 @@ func (c Client) Write(ctx context.Context, tables schema.Tables, res <-chan de`
`79`	`80`	`}`
`80`	`81`	`return fmt.Errorf("failed to execute batch with pgerror: %s: %w", pgErrToStr(pgErr), err)`
`81`	`82`	`}`
`82`		`- atomic.AddUint64(&c.metrics.Writes, uint64(c.batchSize))`
	`83`	`+ atomic.AddUint64(&c.metrics.Writes, uint64(batchSize))`
`83`	`84`	`batch = &pgx.Batch{}`
`84`	`85`	`}`
`85`	`86`	`}`
`86`	`87`
`87`		`- if batch.Len() > 0 {`
	`88`	`+ batchSize := batch.Len()`
	`89`	`+ if batchSize > 0 {`
`88`	`90`	`br := c.conn.SendBatch(ctx, batch)`
`89`	`91`	`if err := br.Close(); err != nil {`
`90`	`92`	`var pgErr *pgconn.PgError`
`@@ -94,7 +96,7 @@ func (c Client) Write(ctx context.Context, tables schema.Tables, res <-chan de`
`94`	`96`	`}`
`95`	`97`	`return fmt.Errorf("failed to execute batch with pgerror: %s: %w", pgErrToStr(pgErr), err)`
`96`	`98`	`}`
`97`		`- atomic.AddUint64(&c.metrics.Writes, uint64(c.batchSize))`
	`99`	`+ atomic.AddUint64(&c.metrics.Writes, uint64(batchSize))`
`98`	`100`	`}`
`99`	`101`
`100`	`102`	`return nil`