Repository: jackc/pgx
Pull request #1079 (2021-09-10 16:39:43)

Add some comments to pgxpool package
Hi @jackc and other maintainers. I am switching from `lib/pq` to this library and it's been a pleasure so far. Thanks a lot for all the hard work! As I'm reading through the library and trying to understand how it works, I thought I could help by adding a few comments. If it's not helpful, I won't continue, but I wanted to see if this was something you're open to. A lot of the time, I'm copying comments from other functions that have the same or similar meaning and tweaking them slightly (to consider the pool, etc.)...

Thanks again!

Dan


Patch:
diff --git a/batch.go b/batch.go
index 4b96ca19..f0479ea6 100644
--- a/batch.go
+++ b/batch.go
@@ -41,6 +41,9 @@ type BatchResults interface {
 	// QueryRow reads the results from the next query in the batch as if the query has been sent with Conn.QueryRow.
 	QueryRow() Row
 
+	// QueryFunc reads the results from the next query in the batch as if the query has been sent with Conn.QueryFunc.
+	QueryFunc(scans []interface{}, f func(QueryFuncRow) error) (pgconn.CommandTag, error)
+
 	// Close closes the batch operation. This must be called before the underlying connection can be used again. Any error
 	// that occurred during a batch operation may have made it impossible to resyncronize the connection with the server.
 	// In this case the underlying connection will have been closed.
@@ -135,6 +138,33 @@ func (br *batchResults) Query() (Rows, error) {
 	return rows, nil
 }
 
+// QueryFunc reads the results from the next query in the batch as if the query has been sent with Conn.QueryFunc.
+func (br *batchResults) QueryFunc(scans []interface{}, f func(QueryFuncRow) error) (pgconn.CommandTag, error) {
+	rows, err := br.Query()
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	for rows.Next() {
+		err = rows.Scan(scans...)
+		if err != nil {
+			return nil, err
+		}
+
+		err = f(rows)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+
+	return rows.CommandTag(), nil
+}
+
 // QueryRow reads the results from the next query in the batch as if the query has been sent with QueryRow.
 func (br *batchResults) QueryRow() Row {
 	rows, _ := br.Query()
diff --git a/pgxpool/batch_results.go b/pgxpool/batch_results.go
index ec393a8c..c625a474 100644
--- a/pgxpool/batch_results.go
+++ b/pgxpool/batch_results.go
@@ -17,6 +17,10 @@ func (br errBatchResults) Query() (pgx.Rows, error) {
 	return errRows{err: br.err}, br.err
 }
 
+func (br errBatchResults) QueryFunc(scans []interface{}, f func(pgx.QueryFuncRow) error) (pgconn.CommandTag, error) {
+	return nil, br.err
+}
+
 func (br errBatchResults) QueryRow() pgx.Row {
 	return errRow{err: br.err}
 }
@@ -38,6 +42,10 @@ func (br *poolBatchResults) Query() (pgx.Rows, error) {
 	return br.br.Query()
 }
 
+func (br *poolBatchResults) QueryFunc(scans []interface{}, f func(pgx.QueryFuncRow) error) (pgconn.CommandTag, error) {
+	return br.br.QueryFunc(scans, f)
+}
+
 func (br *poolBatchResults) QueryRow() pgx.Row {
 	return br.br.QueryRow()
 }
diff --git a/pgxpool/conn.go b/pgxpool/conn.go
index 29ca04d0..0b59d741 100644
--- a/pgxpool/conn.go
+++ b/pgxpool/conn.go
@@ -70,10 +70,12 @@ func (c *Conn) CopyFrom(ctx context.Context, tableName pgx.Identifier, columnNam
 	return c.Conn().CopyFrom(ctx, tableName, columnNames, rowSrc)
 }
 
+// Begin starts a transaction block from the *Conn without explicitly setting a transaction mode (see BeginTx with TxOptions if transaction mode is required).
 func (c *Conn) Begin(ctx context.Context) (pgx.Tx, error) {
 	return c.Conn().Begin(ctx)
 }
 
+// BeginTx starts a transaction block from the *Conn with txOptions determining the transaction mode.
 func (c *Conn) BeginTx(ctx context.Context, txOptions pgx.TxOptions) (pgx.Tx, error) {
 	return c.Conn().BeginTx(ctx, txOptions)
 }
diff --git a/pgxpool/pool.go b/pgxpool/pool.go
index 54982a55..8b015c20 100644
--- a/pgxpool/pool.go
+++ b/pgxpool/pool.go
@@ -375,6 +375,7 @@ func (p *Pool) checkMinConns() {
 	}
 }
 
+// Acquire returns a connection (*Conn) from the Pool
 func (p *Pool) Acquire(ctx context.Context) (*Conn, error) {
 	for {
 		res, err := p.p.Acquire(ctx)
@@ -483,9 +484,18 @@ func (p *Pool) SendBatch(ctx context.Context, b *pgx.Batch) pgx.BatchResults {
 	return &poolBatchResults{br: br, c: c}
 }
 
+// Begin acquires a connection from the Pool and starts a transaction. Unlike database/sql, the context only affects the begin command. i.e. there is no
+// auto-rollback on context cancellation. Begin initiates a transaction block without explicitly setting a transaction mode for the block (see BeginTx with TxOptions if transaction mode is required).
+// *pgxpool.Tx is returned, which implements the pgx.Tx interface.
+// Commit or Rollback must be called on the returned transaction to finalize the transaction block.
 func (p *Pool) Begin(ctx context.Context) (pgx.Tx, error) {
 	return p.BeginTx(ctx, pgx.TxOptions{})
 }
+
+// BeginTx acquires a connection from the Pool and starts a transaction with pgx.TxOptions determining the transaction mode.
+// Unlike database/sql, the context only affects the begin command. i.e. there is no auto-rollback on context cancellation.
+// *pgxpool.Tx is returned, which implements the pgx.Tx interface.
+// Commit or Rollback must be called on the returned transaction to finalize the transaction block.
 func (p *Pool) BeginTx(ctx context.Context, txOptions pgx.TxOptions) (pgx.Tx, error) {
 	c, err := p.Acquire(ctx)
 	if err != nil {
@@ -525,6 +535,8 @@ func (p *Pool) CopyFrom(ctx context.Context, tableName pgx.Identifier, columnNam
 	return c.Conn().CopyFrom(ctx, tableName, columnNames, rowSrc)
 }
 
+// Ping acquires a connection from the Pool and executes an empty sql statement against it.
+// If the sql returns without error, the database Ping is considered successful, otherwise, the error is returned.
 func (p *Pool) Ping(ctx context.Context) error {
 	c, err := p.Acquire(ctx)
 	if err != nil {
diff --git a/pgxpool/tx.go b/pgxpool/tx.go
index e1c980e1..f983b102 100644
--- a/pgxpool/tx.go
+++ b/pgxpool/tx.go
@@ -7,11 +7,13 @@ import (
 	"github.com/jackc/pgx/v4"
 )
 
+// Tx represents a database transaction acquired from a Pool.
 type Tx struct {
 	t pgx.Tx
 	c *Conn
 }
 
+// Begin starts a pseudo nested transaction implemented with a savepoint.
 func (tx *Tx) Begin(ctx context.Context) (pgx.Tx, error) {
 	return tx.t.Begin(ctx)
 }
@@ -20,6 +22,9 @@ func (tx *Tx) BeginFunc(ctx context.Context, f func(pgx.Tx) error) error {
 	return tx.t.BeginFunc(ctx, f)
 }
 
+// Commit commits the transaction and returns the associated connection back to the Pool. Commit will return ErrTxClosed
+// if the Tx is already closed, but is otherwise safe to call multiple times. If the commit fails with a rollback status
+// (e.g. the transaction was already in a broken state) then ErrTxCommitRollback will be returned.
 func (tx *Tx) Commit(ctx context.Context) error {
 	err := tx.t.Commit(ctx)
 	if tx.c != nil {
@@ -29,6 +34,9 @@ func (tx *Tx) Commit(ctx context.Context) error {
 	return err
 }
 
+// Rollback rolls back the transaction and returns the associated connection back to the Pool. Rollback will return ErrTxClosed
+// if the Tx is already closed, but is otherwise safe to call multiple times. Hence, defer tx.Rollback() is safe even if
+// tx.Commit() will be called first in a non-error condition.
 func (tx *Tx) Rollback(ctx context.Context) error {
 	err := tx.t.Rollback(ctx)
 	if tx.c != nil {
diff --git a/tx.go b/tx.go
index 973f2b8e..ad399679 100644
--- a/tx.go
+++ b/tx.go
@@ -10,6 +10,7 @@ import (
 	"github.com/jackc/pgconn"
 )
 
+// TxIsoLevel is the transaction isolation level (serializable, repeatable read, read committed or read uncommitted)
 type TxIsoLevel string
 
 // Transaction isolation levels
@@ -20,6 +21,7 @@ const (
 	ReadUncommitted = TxIsoLevel("read uncommitted")
 )
 
+// TxAccessMode is the transaction access mode (read write or read only)
 type TxAccessMode string
 
 // Transaction access modes
@@ -28,6 +30,7 @@ const (
 	ReadOnly  = TxAccessMode("read only")
 )
 
+// TxDeferrableMode is the transaction deferrable mode (deferrable or not deferrable)
 type TxDeferrableMode string
 
 // Transaction deferrable modes
@@ -36,6 +39,7 @@ const (
 	NotDeferrable = TxDeferrableMode("not deferrable")
 )
 
+// TxOptions are transaction modes within a transaction block
 type TxOptions struct {
 	IsoLevel       TxIsoLevel
 	AccessMode     TxAccessMode


Test patch:
diff --git a/batch_test.go b/batch_test.go
index 0b95cd47..988a1682 100644
--- a/batch_test.go
+++ b/batch_test.go
@@ -32,6 +32,7 @@ func TestConnSendBatch(t *testing.T) {
 	batch.Queue("insert into ledger(description, amount) values($1, $2)", "q2", 2)
 	batch.Queue("insert into ledger(description, amount) values($1, $2)", "q3", 3)
 	batch.Queue("select id, description, amount from ledger order by id")
+	batch.Queue("select id, description, amount from ledger order by id")
 	batch.Queue("select sum(amount) from ledger")
 
 	br := conn.SendBatch(context.Background(), batch)
@@ -60,6 +61,16 @@ func TestConnSendBatch(t *testing.T) {
 		t.Errorf("ct.RowsAffected() => %v, want %v", ct.RowsAffected(), 1)
 	}
 
+	selectFromLedgerExpectedRows := []struct {
+		id          int32
+		description string
+		amount      int32
+	}{
+		{1, "q1", 1},
+		{2, "q2", 2},
+		{3, "q3", 3},
+	}
+
 	rows, err := br.Query()
 	if err != nil {
 		t.Error(err)
@@ -68,62 +79,54 @@ func TestConnSendBatch(t *testing.T) {
 	var id int32
 	var description string
 	var amount int32
-	if !rows.Next() {
-		t.Fatal("expected a row to be available")
-	}
-	if err := rows.Scan(&id, &description, &amount); err != nil {
-		t.Fatal(err)
-	}
-	if id != 1 {
-		t.Errorf("id => %v, want %v", id, 1)
-	}
-	if description != "q1" {
-		t.Errorf("description => %v, want %v", description, "q1")
-	}
-	if amount != 1 {
-		t.Errorf("amount => %v, want %v", amount, 1)
-	}
+	rowCount := 0
 
-	if !rows.Next() {
-		t.Fatal("expected a row to be available")
-	}
-	if err := rows.Scan(&id, &description, &amount); err != nil {
-		t.Fatal(err)
-	}
-	if id != 2 {
-		t.Errorf("id => %v, want %v", id, 2)
-	}
-	if description != "q2" {
-		t.Errorf("description => %v, want %v", description, "q2")
-	}
-	if amount != 2 {
-		t.Errorf("amount => %v, want %v", amount, 2)
-	}
+	for rows.Next() {
+		if rowCount >= len(selectFromLedgerExpectedRows) {
+			t.Fatalf("got too many rows: %d", rowCount)
+		}
 
-	if !rows.Next() {
-		t.Fatal("expected a row to be available")
-	}
-	if err := rows.Scan(&id, &description, &amount); err != nil {
-		t.Fatal(err)
-	}
-	if id != 3 {
-		t.Errorf("id => %v, want %v", id, 3)
-	}
-	if description != "q3" {
-		t.Errorf("description => %v, want %v", description, "q3")
-	}
-	if amount != 3 {
-		t.Errorf("amount => %v, want %v", amount, 3)
-	}
+		if err := rows.Scan(&id, &description, &amount); err != nil {
+			t.Fatalf("row %d: %v", rowCount, err)
+		}
+
+		if id != selectFromLedgerExpectedRows[rowCount].id {
+			t.Errorf("id => %v, want %v", id, selectFromLedgerExpectedRows[rowCount].id)
+		}
+		if description != selectFromLedgerExpectedRows[rowCount].description {
+			t.Errorf("description => %v, want %v", description, selectFromLedgerExpectedRows[rowCount].description)
+		}
+		if amount != selectFromLedgerExpectedRows[rowCount].amount {
+			t.Errorf("amount => %v, want %v", amount, selectFromLedgerExpectedRows[rowCount].amount)
+		}
 
-	if rows.Next() {
-		t.Fatal("did not expect a row to be available")
+		rowCount++
 	}
 
 	if rows.Err() != nil {
 		t.Fatal(rows.Err())
 	}
 
+	rowCount = 0
+	_, err = br.QueryFunc([]interface{}{&id, &description, &amount}, func(pgx.QueryFuncRow) error {
+		if id != selectFromLedgerExpectedRows[rowCount].id {
+			t.Errorf("id => %v, want %v", id, selectFromLedgerExpectedRows[rowCount].id)
+		}
+		if description != selectFromLedgerExpectedRows[rowCount].description {
+			t.Errorf("description => %v, want %v", description, selectFromLedgerExpectedRows[rowCount].description)
+		}
+		if amount != selectFromLedgerExpectedRows[rowCount].amount {
+			t.Errorf("amount => %v, want %v", amount, selectFromLedgerExpectedRows[rowCount].amount)
+		}
+
+		rowCount++
+
+		return nil
+	})
+	if err != nil {
+		t.Error(err)
+	}
+
 	err = br.QueryRow().Scan(&amount)
 	if err != nil {
 		t.Error(err)