pgx

README

YugabyteDB Go Driver

This is a Go Driver based on jackc/pgx, with following additional feature:

Connection load balancing

Users can use this feature in two configurations.

Cluster-aware / Uniform connection load balancing

In the cluster-aware connection load balancing, connections are distributed across all the tservers in the cluster, irrespective of their placements.

To enable the cluster-aware connection load balancing, provide the parameter load_balance with value as either true or any in the connection url or the connection string (DSN style). This section explains the different values for load_balance parameter.

"postgres://username:password@localhost:5433/database_name?load_balance=true"

With this parameter specified in the url, the driver will fetch and maintain the list of tservers from the given endpoint (localhost in above example) available in the YugabyteDB cluster and distribute the connections equally across them.

This list is refreshed every 5 minutes, when a new connection request is received.

Application needs to use the same connection url to create every connection it needs, so that the distribution happens equally.

Topology-aware connection load balancing

With topology-aware connnection load balancing, users can target tservers in specific zones by specifying these zones as topology_keys with values in the format cloudname.regionname.zonename. Multiple zones can be specified as comma separated values.

The connections will be distributed equally with the tservers in these zones.

Note that, you would still need to specify load_balance to one of the 5 allowed values to enable the topology-aware connection load balancing.

"postgres://username:password@localhost:5433/database_name?load_balance=true&topology_keys=cloud1.region1.zone1,cloud1.region1.zone2"

Specifying fallback zones

For topology-aware load balancing, you can now specify fallback placements too. This is not applicable for cluster-aware load balancing. Each placement value can be suffixed with a colon (:) followed by a preference value between 1 and 10. A preference value of :1 means it is a primary placement. A preference value of :2 means it is the first fallback placement and so on.If no preference value is provided, it is considered to be a primary placement (equivalent to one with preference value :1). Example given below.

"postgres://username:password@localhost:5433/database_name?load_balance=any&topology_keys=cloud1.region1.zone1:1,cloud1.region1.zone2:2";

You can also use * for specifying all the zones in a given region as shown below. This is not allowed for cloud or region values.

"postgres://username:password@localhost:5433/database_name?load_balance=any&topology_keys=cloud1.region1.*:1,cloud1.region2.*:2";

The driver attempts to connect to a node in following order: the least loaded node in the 1) primary placement(s), else in the 2) first fallback if specified, else in the 3) second fallback if specified and so on. If no nodes are available either in primary placement(s) or in any of the fallback placements, then nodes in the rest of the cluster are attempted. And this repeats for each connection request.

Specifying Refresh Interval

Users can specify Refresh Time Interval, in seconds. It is the time interval between two attempts to refresh the information about cluster nodes. Default is 300. Valid values are integers between 0 and 600. Value 0 means refresh for each connection request. Any value outside this range is ignored and the default is used.

To specify Refresh Interval, use the parameter yb_servers_refresh_interval in the connection url or the connection string.

"postgres://username:password@localhost:5433/database_name?yb_servers_refresh_interval=X&load_balance=true&topology_keys=cloud1.region1.*:1,cloud1.region2.*:2";

Same parameters can be specified in the connection url while using the pgxpool.Connect() API.

Other Connection Parameters:

fallback_to_topology_keys_only

Applicable only for TopologyAware Load Balancing. When set to true, the smart driver does not attempt to connect to servers outside of primary and fallback placements specified via property. The default behaviour is to fallback to any available server in the entire cluster.(default value: false)

failed_host_reconnect_delay_secs

The driver marks a server as failed with a timestamp, when it cannot connect to it. Later, whenever it refreshes the server list via yb_servers(), if it sees the failed server in the response, it marks the server as UP only if failed-host-reconnect-delay-secs time has elapsed. (The yb_servers() function does not remove a failed server immediately from its result and retains it for a while.)(default value: 5 seconds)

Read Replica Cluster

PGX smart driver also enables load balancing across nodes in primary clusters which have associated Read Replica cluster.

The connection property load-balance allows five values using which users can distribute connections among different combination of nodes as per their requirements:

• only-rr - Create connections only on Read Replica nodes
• only-primary - Create connections only on primary cluster nodes
• prefer-rr - Create connections on Read Replica nodes. If none available, on any node in the cluster including primary cluster nodes
• prefer-primary - Create connections on primary cluster nodes. If none available, on any node in the cluster including Read Replica nodes
• any or true - Equivalent to value true. Create connections on any node in the primary or Read Replica cluster

default value is false

Details about the upstream pgx driver - which hold true for this driver as well - are given below.

pgx - PostgreSQL Driver and Toolkit

pgx is a pure Go driver and toolkit for PostgreSQL.

The pgx driver is a low-level, high performance interface. It also includes an adapter for the standard database/sql interface.

The toolkit component is a related set of packages that implement PostgreSQL functionality such as parsing the wire protocol and type mapping between PostgreSQL and Go. These underlying packages can be used to implement alternative drivers, proxies, load balancers, logical replication clients, etc.

Example Usage

package main

import (
	"context"
	"fmt"
	"os"

	"github.com/yugabyte/pgx/v5"
)

func main() {
	// urlExample := "postgres://username:password@localhost:5433/database_name"
	conn, err := pgx.Connect(context.Background(), os.Getenv("DATABASE_URL"))
	if err != nil {
		fmt.Fprintf(os.Stderr, "Unable to connect to database: %v\n", err)
		os.Exit(1)
	}
	defer conn.Close(context.Background())

	var name string
	var weight int64
	err = conn.QueryRow(context.Background(), "select name, weight from widgets where id=$1", 42).Scan(&name, &weight)
	if err != nil {
		fmt.Fprintf(os.Stderr, "QueryRow failed: %v\n", err)
		os.Exit(1)
	}

	fmt.Println(name, weight)
}

See the getting started guide for more information.

Features

• Support for approximately 70 different PostgreSQL types
• Automatic statement preparation and caching
• Batch queries
• Single-round trip query mode
• Full TLS connection control
• Binary format support for custom types (allows for much quicker encoding/decoding)
• COPY protocol support for faster bulk data loads
• Tracing and logging support
• Connection pool with after-connect hook for arbitrary connection setup
• Conversion of PostgreSQL arrays to Go slice mappings for integers, floats, and strings
• hstore support
• json and jsonb support
• Maps inet and cidr PostgreSQL types to netip.Addr and netip.Prefix
• Large object support
• NULL mapping to pointer to pointer
• Supports database/sql.Scanner and database/sql/driver.Valuer interfaces for custom types
• Notice response handling
• Simulated nested transactions with savepoints

Choosing Between the pgx and database/sql Interfaces

The pgx interface is faster.

The pgx interface is recommended when:

1. The application only targets PostgreSQL.
2. No other libraries that require database/sql are in use.

It is also possible to use the database/sql interface and convert a connection to the lower-level pgx interface as needed.

Testing

See CONTRIBUTING.md for setup instructions.

Architecture

See the presentation at Golang Estonia, PGX Top to Bottom for a description of pgx architecture.

Supported Go and PostgreSQL Versions

pgx supports the same versions of Go and PostgreSQL that are supported by their respective teams. For Go that is the two most recent major releases and for PostgreSQL the major releases in the last 5 years. This means pgx supports Go 1.20 and higher and PostgreSQL 12 and higher.

Version Policy

pgx follows semantic versioning for the documented public API on stable releases. v5 is the latest stable major version.

PGX Family Libraries

github.com/jackc/pglogrepl

pglogrepl provides functionality to act as a client for PostgreSQL logical replication.

pgx supports the same versions of Go and PostgreSQL that are supported by their respective teams. For Go that is the two most recent major releases and for PostgreSQL the major releases in the last 5 years. This means pgx supports Go 1.16 and higher and PostgreSQL 10 and higher.

pgmock offers the ability to create a server that mocks the PostgreSQL wire protocol. This is used internally to test pgx by purposely inducing unusual errors. pgproto3 and pgmock together provide most of the foundational tooling required to implement a PostgreSQL proxy or MitM (such as for a custom connection pooler).

github.com/jackc/tern

tern is a stand-alone SQL migration system.

github.com/jackc/pgerrcode

pgerrcode contains constants for the PostgreSQL error codes.

Adapters for 3rd Party Types

github.com/yugabyte/pgx/v4/pgxpool

github.com/yugabyte/pgx/v4/stdlib

• https://github.com/jackhopner/pgx-xray-tracer

Adapters for 3rd Party Loggers

These adapters can be used with the tracelog package.

• github.com/jackc/pgx-go-kit-log
• github.com/jackc/pgx-log15
• github.com/jackc/pgx-logrus
• github.com/jackc/pgx-zap
• github.com/jackc/pgx-zerolog
• github.com/mcosta74/pgx-slog
• github.com/kataras/pgx-golog

3rd Party Libraries with PGX Support

github.com/pashagolub/pgxmock

pgxmock is a mock library implementing pgx interfaces. pgxmock has one and only purpose - to simulate pgx behavior in tests, without needing a real database connection.

github.com/georgysavva/scany

Library for scanning data from a database into Go structs and more.

github.com/vingarcia/ksql

A carefully designed SQL client for making using SQL easier, more productive, and less error-prone on Golang.

https://github.com/otan/gopgkrb5

Adds GSSAPI / Kerberos authentication support.

github.com/wcamarao/pmx

Explicit data mapping and scanning library for Go structs and slices.

github.com/stephenafamo/scan

Type safe and flexible package for scanning database data into Go types. Supports, structs, maps, slices and custom mapping functions.

https://github.com/z0ne-dev/mgx

Code first migration library for native pgx (no database/sql abstraction).

Documentation

Overview

Package pgx is a PostgreSQL database driver.

pgx provides a native PostgreSQL driver and can act as a database/sql driver. The native PostgreSQL interface is similar to the database/sql interface while providing better speed and access to PostgreSQL specific features. Use github.com/yugabyte/pgx/v5/stdlib to use pgx as a database/sql compatible driver. See that package's documentation for details.

Establishing a Connection

The primary way of establishing a connection is with pgx.Connect:

conn, err := pgx.Connect(context.Background(), os.Getenv("DATABASE_URL"))

The database connection string can be in URL or DSN format. Both PostgreSQL settings and pgx settings can be specified here. In addition, a config struct can be created by ParseConfig and modified before establishing the connection with ConnectConfig to configure settings such as tracing that cannot be configured with a connection string.

Connection Pool

*pgx.Conn represents a single connection to the database and is not concurrency safe. Use package github.com/yugabyte/pgx/v5/pgxpool for a concurrency safe connection pool.

Query Interface

pgx implements Query in the familiar database/sql style. However, pgx provides generic functions such as CollectRows and ForEachRow that are a simpler and safer way of processing rows than manually calling rows.Next(), rows.Scan, and rows.Err().

CollectRows can be used collect all returned rows into a slice.

rows, _ := conn.Query(context.Background(), "select generate_series(1,$1)", 5)
numbers, err := pgx.CollectRows(rows, pgx.RowTo[int32])
if err != nil {
  return err
}
// numbers => [1 2 3 4 5]

ForEachRow can be used to execute a callback function for every row. This is often easier than iterating over rows directly.

var sum, n int32
rows, _ := conn.Query(context.Background(), "select generate_series(1,$1)", 10)
_, err := pgx.ForEachRow(rows, []any{&n}, func() error {
  sum += n
  return nil
})
if err != nil {
  return err
}

pgx also implements QueryRow in the same style as database/sql.

var name string
var weight int64
err := conn.QueryRow(context.Background(), "select name, weight from widgets where id=$1", 42).Scan(&name, &weight)
if err != nil {
    return err
}

Use Exec to execute a query that does not return a result set.

commandTag, err := conn.Exec(context.Background(), "delete from widgets where id=$1", 42)
if err != nil {
    return err
}
if commandTag.RowsAffected() != 1 {
    return errors.New("No row found to delete")
}

PostgreSQL Data Types

pgx uses the pgtype package to converting Go values to and from PostgreSQL values. It supports many PostgreSQL types directly and is customizable and extendable. User defined data types such as enums, domains, and composite types may require type registration. See that package's documentation for details.

Transactions

Transactions are started by calling Begin.

tx, err := conn.Begin(context.Background())
if err != nil {
    return err
}
// Rollback is safe to call even if the tx is already closed, so if
// the tx commits successfully, this is a no-op
defer tx.Rollback(context.Background())

_, err = tx.Exec(context.Background(), "insert into foo(id) values (1)")
if err != nil {
    return err
}

err = tx.Commit(context.Background())
if err != nil {
    return err
}

The Tx returned from Begin also implements the Begin method. This can be used to implement pseudo nested transactions. These are internally implemented with savepoints.

Use BeginTx to control the transaction mode. BeginTx also can be used to ensure a new transaction is created instead of a pseudo nested transaction.

BeginFunc and BeginTxFunc are functions that begin a transaction, execute a function, and commit or rollback the transaction depending on the return value of the function. These can be simpler and less error prone to use.

err = pgx.BeginFunc(context.Background(), conn, func(tx pgx.Tx) error {
    _, err := tx.Exec(context.Background(), "insert into foo(id) values (1)")
    return err
})
if err != nil {
    return err
}

Prepared Statements

Prepared statements can be manually created with the Prepare method. However, this is rarely necessary because pgx includes an automatic statement cache by default. Queries run through the normal Query, QueryRow, and Exec functions are automatically prepared on first execution and the prepared statement is reused on subsequent executions. See ParseConfig for information on how to customize or disable the statement cache.

Copy Protocol

Use CopyFrom to efficiently insert multiple rows at a time using the PostgreSQL copy protocol. CopyFrom accepts a CopyFromSource interface. If the data is already in a [][]any use CopyFromRows to wrap it in a CopyFromSource interface. Or implement CopyFromSource to avoid buffering the entire data set in memory.

rows := [][]any{
    {"John", "Smith", int32(36)},
    {"Jane", "Doe", int32(29)},
}

copyCount, err := conn.CopyFrom(
    context.Background(),
    pgx.Identifier{"people"},
    []string{"first_name", "last_name", "age"},
    pgx.CopyFromRows(rows),
)

When you already have a typed array using CopyFromSlice can be more convenient.

rows := []User{
    {"John", "Smith", 36},
    {"Jane", "Doe", 29},
}

copyCount, err := conn.CopyFrom(
    context.Background(),
    pgx.Identifier{"people"},
    []string{"first_name", "last_name", "age"},
    pgx.CopyFromSlice(len(rows), func(i int) ([]any, error) {
        return []any{rows[i].FirstName, rows[i].LastName, rows[i].Age}, nil
    }),
)

CopyFrom can be faster than an insert with as few as 5 rows.

Tracing and Logging

pgx supports tracing by setting ConnConfig.Tracer.

In addition, the tracelog package provides the TraceLog type which lets a traditional logger act as a Tracer.

For debug tracing of the actual PostgreSQL wire protocol messages see github.com/yugabyte/pgx/v5/pgproto3.

Lower Level PostgreSQL Functionality

github.com/yugabyte/pgx/v5/pgconn contains a lower level PostgreSQL driver roughly at the level of libpq. pgx.Conn in implemented on top of pgconn. The Conn.PgConn() method can be used to access this lower layer.

PgBouncer

By default pgx automatically uses prepared statements. Prepared statements are incompatible with PgBouncer. This can be disabled by setting a different QueryExecMode in ConnConfig.DefaultQueryExecMode.

Index

• Constants
• Variables
• func AppendRows[T any, S ~[]T](slice S, rows Rows, fn RowToFunc[T]) (S, error)
• func BeginFunc(ctx context.Context, db interface{ ... }, fn func(Tx) error) (err error)
• func BeginTxFunc(ctx context.Context, db interface{ ... }, txOptions TxOptions, ...) (err error)
• func CollectExactlyOneRow[T any](rows Rows, fn RowToFunc[T]) (T, error)
• func CollectOneRow[T any](rows Rows, fn RowToFunc[T]) (T, error)
• func CollectRows[T any](rows Rows, fn RowToFunc[T]) ([]T, error)
• func EmptyHostLoad() map[string]map[string]int
• func ForEachRow(rows Rows, scans []any, fn func() error) (pgconn.CommandTag, error)
• func GetAZInfo() map[string]map[string][]string
• func GetHostLoad() map[string]map[string]int
• func LookupIP(host string) string
• func RowTo[T any](row CollectableRow) (T, error)
• func RowToAddrOf[T any](row CollectableRow) (*T, error)
• func RowToAddrOfStructByName[T any](row CollectableRow) (*T, error)
• func RowToAddrOfStructByNameLax[T any](row CollectableRow) (*T, error)
• func RowToAddrOfStructByPos[T any](row CollectableRow) (*T, error)
• func RowToMap(row CollectableRow) (map[string]any, error)
• func RowToStructByName[T any](row CollectableRow) (T, error)
• func RowToStructByNameLax[T any](row CollectableRow) (T, error)
• func RowToStructByPos[T any](row CollectableRow) (T, error)
• func ScanRow(typeMap *pgtype.Map, fieldDescriptions []pgconn.FieldDescription, ...) error
• type Batch
  • func (b *Batch) Len() int
  • func (b *Batch) Queue(query string, arguments ...any) *QueuedQuery
• type BatchResults
• type BatchTracer
• type ClusterLoadInfo
  • func NewClusterLoadInfo(ctx context.Context, config *ConnConfig) *ClusterLoadInfo
• type CollectableRow
• type Conn
  • func Connect(ctx context.Context, connString string) (*Conn, error)
  • func ConnectConfig(ctx context.Context, connConfig *ConnConfig) (*Conn, error)
  • func ConnectWithOptions(ctx context.Context, connString string, options ParseConfigOptions) (*Conn, error)
  • func (c *Conn) Begin(ctx context.Context) (Tx, error)
  • func (c *Conn) BeginTx(ctx context.Context, txOptions TxOptions) (Tx, error)
  • func (c *Conn) Close(ctx context.Context) error
  • func (c *Conn) Config() *ConnConfig
  • func (c *Conn) CopyFrom(ctx context.Context, tableName Identifier, columnNames []string, ...) (int64, error)
  • func (c *Conn) Deallocate(ctx context.Context, name string) error
  • func (c *Conn) DeallocateAll(ctx context.Context) error
  • func (c *Conn) Exec(ctx context.Context, sql string, arguments ...any) (pgconn.CommandTag, error)
  • func (c *Conn) IsClosed() bool
  • func (c *Conn) LoadType(ctx context.Context, typeName string) (*pgtype.Type, error)
  • func (c *Conn) PgConn() *pgconn.PgConn
  • func (c *Conn) Ping(ctx context.Context) error
  • func (c *Conn) Prepare(ctx context.Context, name, sql string) (sd *pgconn.StatementDescription, err error)
  • func (c *Conn) Query(ctx context.Context, sql string, args ...any) (Rows, error)
  • func (c *Conn) QueryRow(ctx context.Context, sql string, args ...any) Row
  • func (c *Conn) SendBatch(ctx context.Context, b *Batch) (br BatchResults)
  • func (c *Conn) TypeMap() *pgtype.Map
  • func (c *Conn) WaitForNotification(ctx context.Context) (*pgconn.Notification, error)
• type ConnConfig
  • func ParseConfig(connString string) (*ConnConfig, error)
  • func ParseConfigWithOptions(connString string, options ParseConfigOptions) (*ConnConfig, error)
  • func (cc *ConnConfig) ConnString() string
  • func (cc *ConnConfig) Copy() *ConnConfig
• type ConnectTracer
• type CopyFromSource
  • func CopyFromFunc(nxtf func() (row []any, err error)) CopyFromSource
  • func CopyFromRows(rows [][]any) CopyFromSource
  • func CopyFromSlice(length int, next func(int) ([]any, error)) CopyFromSource
• type CopyFromTracer
• type ExtendedQueryBuilder
  • func (eqb *ExtendedQueryBuilder) Build(m *pgtype.Map, sd *pgconn.StatementDescription, args []any) error
• type Identifier
  • func (ident Identifier) Sanitize() string
• type LargeObject
  • func (o *LargeObject) Close() error
  • func (o *LargeObject) Read(p []byte) (int, error)
  • func (o *LargeObject) Seek(offset int64, whence int) (n int64, err error)
  • func (o *LargeObject) Tell() (n int64, err error)
  • func (o *LargeObject) Truncate(size int64) (err error)
  • func (o *LargeObject) Write(p []byte) (int, error)
• type LargeObjectMode
• type LargeObjects
  • func (o *LargeObjects) Create(ctx context.Context, oid uint32) (uint32, error)
  • func (o *LargeObjects) Open(ctx context.Context, oid uint32, mode LargeObjectMode) (*LargeObject, error)
  • func (o *LargeObjects) Unlink(ctx context.Context, oid uint32) error
• type NamedArgs
  • func (na NamedArgs) RewriteQuery(ctx context.Context, conn *Conn, sql string, args []any) (newSQL string, newArgs []any, err error)
• type ParseConfigOptions
• type PrepareTracer
• type QueryExecMode
  • func (m QueryExecMode) String() string
• type QueryResultFormats
• type QueryResultFormatsByOID
• type QueryRewriter
• type QueryTracer
• type QueuedQuery
  • func (qq *QueuedQuery) Exec(fn func(ct pgconn.CommandTag) error)
  • func (qq *QueuedQuery) Query(fn func(rows Rows) error)
  • func (qq *QueuedQuery) QueryRow(fn func(row Row) error)
• type Row
• type RowScanner
• type RowToFunc
• type Rows
  • func RowsFromResultReader(typeMap *pgtype.Map, resultReader *pgconn.ResultReader) Rows
• type ScanArgError
  • func (e ScanArgError) Error() string
  • func (e ScanArgError) Unwrap() error
• type TraceBatchEndData
• type TraceBatchQueryData
• type TraceBatchStartData
• type TraceConnectEndData
• type TraceConnectStartData
• type TraceCopyFromEndData
• type TraceCopyFromStartData
• type TracePrepareEndData
• type TracePrepareStartData
• type TraceQueryEndData
• type TraceQueryStartData
• type Tx
• type TxAccessMode
• type TxDeferrableMode
• type TxIsoLevel
• type TxOptions

Examples

• CollectRows
• Conn.Query
• Conn.SendBatch
• ForEachRow
• RowTo
• RowToAddrOf
• RowToStructByName
• RowToStructByNameLax
• RowToStructByPos

Constants

const (
	TextFormatCode   = 0
	BinaryFormatCode = 1
)

PostgreSQL format codes

const CONTROL_CONN_TIMEOUT = 15 * time.Second

const DECREMENT_COUNT byte = 5

Indicate to the Go routine processing the requestChan that it should decrease the connection count for the given host by one

const DEFAULT_FAILED_HOST_RECONNECT_DELAY_SECS = 5

const GET_LB_CONN byte = 4

Indicate to the Go routine processing the requestChan that it should return a least loaded tserver host, port

const HOSTS_EXHAUSTED byte = 3

Both the addresses (host, public_ip) of tserver to be tried, but no success with private addresses to create a connection

const LB_QUERY = "SELECT host,port,num_connections,node_type,cloud,region,zone,public_ip FROM yb_servers()"

const MAX_FAILED_HOST_RECONNECT_DELAY_SECS = 60

const MAX_INTERVAL_SECONDS = 600

const MAX_PREFERENCE_VALUE = 10

const MAX_RETRIES = 20

const NO_SERVERS_MSG = "could not find a server to connect to"

const REFRESH_INTERVAL_SECONDS = 300

const TRY_HOSTS_PUBLIC_IP byte = 2

Try both the addresses (host, public_ip) of tservers to create a connection

const USE_HOSTS byte = 0

-- Values for ClusterLoadInfo.flags -- Use private address (host) of tservers to create a connection

const USE_PUBLIC_IP byte = 1

Use public address (public_ip) of tservers to create a connection

Variables

var (
	// ErrNoRows occurs when rows are expected but none are returned.
	ErrNoRows = errors.New("no rows in result set")
	// ErrTooManyRows occurs when more rows than expected are returned.
	ErrTooManyRows = errors.New("too many rows in result set")
)

var ErrFallbackToOriginalBehaviour = errors.New("no preferred server available, fallback-to-topology-keys-only is set to true")

var ErrTxClosed = errors.New("tx is closed")

var ErrTxCommitRollback = errors.New("commit unexpectedly resulted in rollback")

ErrTxCommitRollback occurs when an error has occurred in a transaction and Commit() is called. PostgreSQL accepts COMMIT on aborted transactions, but it is treated as ROLLBACK.

Functions

func AppendRows

func AppendRows[T any, S ~[]T](slice S, rows Rows, fn RowToFunc[T]) (S, error)

AppendRows iterates through rows, calling fn for each row, and appending the results into a slice of T.

func BeginFunc

func BeginFunc(
	ctx context.Context,
	db interface {
		Begin(ctx context.Context) (Tx, error)
	},
	fn func(Tx) error,
) (err error)

BeginFunc calls Begin on db and then calls fn. If fn does not return an error then it calls Commit on db. If fn returns an error it calls Rollback on db. The context will be used when executing the transaction control statements (BEGIN, ROLLBACK, and COMMIT) but does not otherwise affect the execution of fn.

func BeginTxFunc

func BeginTxFunc(
	ctx context.Context,
	db interface {
		BeginTx(ctx context.Context, txOptions TxOptions) (Tx, error)
	},
	txOptions TxOptions,
	fn func(Tx) error,
) (err error)

BeginTxFunc calls BeginTx on db and then calls fn. If fn does not return an error then it calls Commit on db. If fn returns an error it calls Rollback on db. The context will be used when executing the transaction control statements (BEGIN, ROLLBACK, and COMMIT) but does not otherwise affect the execution of fn.

func CollectExactlyOneRow

func CollectExactlyOneRow[T any](rows Rows, fn RowToFunc[T]) (T, error)

CollectExactlyOneRow calls fn for the first row in rows and returns the result.

• If no rows are found returns an error where errors.Is(ErrNoRows) is true.
• If more than 1 row is found returns an error where errors.Is(ErrTooManyRows) is true.

func CollectOneRow

func CollectOneRow[T any](rows Rows, fn RowToFunc[T]) (T, error)

CollectOneRow calls fn for the first row in rows and returns the result. If no rows are found returns an error where errors.Is(ErrNoRows) is true. CollectOneRow is to CollectRows as QueryRow is to Query.

func CollectRows

func CollectRows[T any](rows Rows, fn RowToFunc[T]) ([]T, error)

CollectRows iterates through rows, calling fn for each row, and collecting the results into a slice of T.

Example

This example uses CollectRows with a manually written collector function. In most cases RowTo, RowToAddrOf, RowToStructByPos, RowToAddrOfStructByPos, or another generic function would be used.

ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second)
defer cancel()

conn, err := pgx.Connect(ctx, os.Getenv("PGX_TEST_DATABASE"))
if err != nil {
	fmt.Printf("Unable to establish connection: %v", err)
	return
}

rows, _ := conn.Query(ctx, `select n from generate_series(1, 5) n`)
numbers, err := pgx.CollectRows(rows, func(row pgx.CollectableRow) (int32, error) {
	var n int32
	err := row.Scan(&n)
	return n, err
})
if err != nil {
	fmt.Printf("CollectRows error: %v", err)
	return
}

fmt.Println(numbers)

Output:

[1 2 3 4 5]

func EmptyHostLoad

func EmptyHostLoad() map[string]map[string]int

For test purpose

func ForEachRow

func ForEachRow(rows Rows, scans []any, fn func() error) (pgconn.CommandTag, error)

ForEachRow iterates through rows. For each row it scans into the elements of scans and calls fn. If any row fails to scan or fn returns an error the query will be aborted and the error will be returned. Rows will be closed when ForEachRow returns.

Example

conn, err := pgx.Connect(context.Background(), os.Getenv("PGX_TEST_DATABASE"))
if err != nil {
	fmt.Printf("Unable to establish connection: %v", err)
	return
}

rows, _ := conn.Query(
	context.Background(),
	"select n, n * 2 from generate_series(1, $1) n",
	3,
)
var a, b int
_, err = pgx.ForEachRow(rows, []any{&a, &b}, func() error {
	fmt.Printf("%v, %v\n", a, b)
	return nil
})
if err != nil {
	fmt.Printf("ForEachRow error: %v", err)
	return
}

Output:

1, 2
2, 4
3, 6

func GetAZInfo

func GetAZInfo() map[string]map[string][]string

For test purpose

func GetHostLoad

func GetHostLoad() map[string]map[string]int

For test purpose

func LookupIP

func LookupIP(host string) string

func RowTo

func RowTo[T any](row CollectableRow) (T, error)

RowTo returns a T scanned from row.

Example

ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second)
defer cancel()

conn, err := pgx.Connect(ctx, os.Getenv("PGX_TEST_DATABASE"))
if err != nil {
	fmt.Printf("Unable to establish connection: %v", err)
	return
}

rows, _ := conn.Query(ctx, `select n from generate_series(1, 5) n`)
numbers, err := pgx.CollectRows(rows, pgx.RowTo[int32])
if err != nil {
	fmt.Printf("CollectRows error: %v", err)
	return
}

fmt.Println(numbers)

Output:

[1 2 3 4 5]

func RowToAddrOf

func RowToAddrOf[T any](row CollectableRow) (*T, error)

RowTo returns a the address of a T scanned from row.

Example

ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second)
defer cancel()

conn, err := pgx.Connect(ctx, os.Getenv("PGX_TEST_DATABASE"))
if err != nil {
	fmt.Printf("Unable to establish connection: %v", err)
	return
}

rows, _ := conn.Query(ctx, `select n from generate_series(1, 5) n`)
pNumbers, err := pgx.CollectRows(rows, pgx.RowToAddrOf[int32])
if err != nil {
	fmt.Printf("CollectRows error: %v", err)
	return
}

for _, p := range pNumbers {
	fmt.Println(*p)
}

Output:

1
2
3
4
5

func RowToAddrOfStructByName

func RowToAddrOfStructByName[T any](row CollectableRow) (*T, error)

RowToAddrOfStructByName returns the address of a T scanned from row. T must be a struct. T must have the same number of named public fields as row has fields. The row and T fields will be matched by name. The match is case-insensitive. The database column name can be overridden with a "db" struct tag. If the "db" struct tag is "-" then the field will be ignored.

func RowToAddrOfStructByNameLax

func RowToAddrOfStructByNameLax[T any](row CollectableRow) (*T, error)

RowToAddrOfStructByNameLax returns the address of a T scanned from row. T must be a struct. T must have greater than or equal number of named public fields as row has fields. The row and T fields will be matched by name. The match is case-insensitive. The database column name can be overridden with a "db" struct tag. If the "db" struct tag is "-" then the field will be ignored.

func RowToAddrOfStructByPos

func RowToAddrOfStructByPos[T any](row CollectableRow) (*T, error)

RowToAddrOfStructByPos returns the address of a T scanned from row. T must be a struct. T must have the same number a public fields as row has fields. The row and T fields will be matched by position. If the "db" struct tag is "-" then the field will be ignored.

func RowToMap

func RowToMap(row CollectableRow) (map[string]any, error)

RowToMap returns a map scanned from row.

func RowToStructByName

func RowToStructByName[T any](row CollectableRow) (T, error)

RowToStructByName returns a T scanned from row. T must be a struct. T must have the same number of named public fields as row has fields. The row and T fields will be matched by name. The match is case-insensitive. The database column name can be overridden with a "db" struct tag. If the "db" struct tag is "-" then the field will be ignored.

Example

ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second)
defer cancel()

conn, err := pgx.Connect(ctx, os.Getenv("PGX_TEST_DATABASE"))
if err != nil {
	fmt.Printf("Unable to establish connection: %v", err)
	return
}

if conn.PgConn().ParameterStatus("crdb_version") != "" {
	// Skip test / example when running on CockroachDB. Since an example can't be skipped fake success instead.
	fmt.Println(`Cheeseburger: $10
Fries: $5
Soft Drink: $3`)
	return
}

// Setup example schema and data.
_, err = conn.Exec(ctx, `
create temporary table products (
	id int primary key generated by default as identity,
	name varchar(100) not null,
	price int not null
);

insert into products (name, price) values
	('Cheeseburger', 10),
	('Double Cheeseburger', 14),
	('Fries', 5),
	('Soft Drink', 3);
`)
if err != nil {
	fmt.Printf("Unable to setup example schema and data: %v", err)
	return
}

type product struct {
	ID    int32
	Name  string
	Price int32
}

rows, _ := conn.Query(ctx, "select * from products where price < $1 order by price desc", 12)
products, err := pgx.CollectRows(rows, pgx.RowToStructByName[product])
if err != nil {
	fmt.Printf("CollectRows error: %v", err)
	return
}

for _, p := range products {
	fmt.Printf("%s: $%d\n", p.Name, p.Price)
}

Output:

Cheeseburger: $10
Fries: $5
Soft Drink: $3

func RowToStructByNameLax

func RowToStructByNameLax[T any](row CollectableRow) (T, error)

RowToStructByNameLax returns a T scanned from row. T must be a struct. T must have greater than or equal number of named public fields as row has fields. The row and T fields will be matched by name. The match is case-insensitive. The database column name can be overridden with a "db" struct tag. If the "db" struct tag is "-" then the field will be ignored.

Example

ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second)
defer cancel()

conn, err := pgx.Connect(ctx, os.Getenv("PGX_TEST_DATABASE"))
if err != nil {
	fmt.Printf("Unable to establish connection: %v", err)
	return
}

if conn.PgConn().ParameterStatus("crdb_version") != "" {
	// Skip test / example when running on CockroachDB. Since an example can't be skipped fake success instead.
	fmt.Println(`Cheeseburger: $10
Fries: $5
Soft Drink: $3`)
	return
}

// Setup example schema and data.
_, err = conn.Exec(ctx, `
create temporary table products (
	id int primary key generated by default as identity,
	name varchar(100) not null,
	price int not null
);

insert into products (name, price) values
	('Cheeseburger', 10),
	('Double Cheeseburger', 14),
	('Fries', 5),
	('Soft Drink', 3);
`)
if err != nil {
	fmt.Printf("Unable to setup example schema and data: %v", err)
	return
}

type product struct {
	ID    int32
	Name  string
	Type  string
	Price int32
}

rows, _ := conn.Query(ctx, "select * from products where price < $1 order by price desc", 12)
products, err := pgx.CollectRows(rows, pgx.RowToStructByNameLax[product])
if err != nil {
	fmt.Printf("CollectRows error: %v", err)
	return
}

for _, p := range products {
	fmt.Printf("%s: $%d\n", p.Name, p.Price)
}

Output:

Cheeseburger: $10
Fries: $5
Soft Drink: $3

func RowToStructByPos

func RowToStructByPos[T any](row CollectableRow) (T, error)

RowToStructByPos returns a T scanned from row. T must be a struct. T must have the same number a public fields as row has fields. The row and T fields will be matched by position. If the "db" struct tag is "-" then the field will be ignored.

Example

ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second)
defer cancel()

conn, err := pgx.Connect(ctx, os.Getenv("PGX_TEST_DATABASE"))
if err != nil {
	fmt.Printf("Unable to establish connection: %v", err)
	return
}

if conn.PgConn().ParameterStatus("crdb_version") != "" {
	// Skip test / example when running on CockroachDB. Since an example can't be skipped fake success instead.
	fmt.Println(`Cheeseburger: $10
Fries: $5
Soft Drink: $3`)
	return
}

// Setup example schema and data.
_, err = conn.Exec(ctx, `
create temporary table products (
	id int primary key generated by default as identity,
	name varchar(100) not null,
	price int not null
);

insert into products (name, price) values
	('Cheeseburger', 10),
	('Double Cheeseburger', 14),
	('Fries', 5),
	('Soft Drink', 3);
`)
if err != nil {
	fmt.Printf("Unable to setup example schema and data: %v", err)
	return
}

type product struct {
	ID    int32
	Name  string
	Price int32
}

rows, _ := conn.Query(ctx, "select * from products where price < $1 order by price desc", 12)
products, err := pgx.CollectRows(rows, pgx.RowToStructByPos[product])
if err != nil {
	fmt.Printf("CollectRows error: %v", err)
	return
}

for _, p := range products {
	fmt.Printf("%s: $%d\n", p.Name, p.Price)
}

Output:

Cheeseburger: $10
Fries: $5
Soft Drink: $3

func ScanRow

func ScanRow(typeMap *pgtype.Map, fieldDescriptions []pgconn.FieldDescription, values [][]byte, dest ...any) error

ScanRow decodes raw row data into dest. It can be used to scan rows read from the lower level pgconn interface.

typeMap - OID to Go type mapping. fieldDescriptions - OID and format of values values - the raw data as returned from the PostgreSQL server dest - the destination that values will be decoded into

Types

type Batch

type Batch struct {
	QueuedQueries []*QueuedQuery
}

Batch queries are a way of bundling multiple queries together to avoid unnecessary network round trips. A Batch must only be sent once.

func (*Batch) Len

func (b *Batch) Len() int

Len returns number of queries that have been queued so far.

func (*Batch) Queue

func (b *Batch) Queue(query string, arguments ...any) *QueuedQuery

Queue queues a query to batch b. query can be an SQL query or the name of a prepared statement. The only pgx option argument that is supported is QueryRewriter. Queries are executed using the connection's DefaultQueryExecMode.

type BatchResults

type BatchResults interface {
	// Exec reads the results from the next query in the batch as if the query has been sent with Conn.Exec. Prefer
	// calling Exec on the QueuedQuery.
	Exec() (pgconn.CommandTag, error)

	// Query reads the results from the next query in the batch as if the query has been sent with Conn.Query. Prefer
	// calling Query on the QueuedQuery.
	Query() (Rows, error)

	// QueryRow reads the results from the next query in the batch as if the query has been sent with Conn.QueryRow.
	// Prefer calling QueryRow on the QueuedQuery.
	QueryRow() Row

	// Close closes the batch operation. All unread results are read and any callback functions registered with
	// QueuedQuery.Query, QueuedQuery.QueryRow, or QueuedQuery.Exec will be called. If a callback function returns an
	// error or the batch encounters an error subsequent callback functions will not be called.
	//
	// Close 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.
	//
	// Close is safe to call multiple times. If it returns an error subsequent calls will return the same error. Callback
	// functions will not be rerun.
	Close() error
}

type BatchTracer

type BatchTracer interface {
	// TraceBatchStart is called at the beginning of SendBatch calls. The returned context is used for the
	// rest of the call and will be passed to TraceBatchQuery and TraceBatchEnd.
	TraceBatchStart(ctx context.Context, conn *Conn, data TraceBatchStartData) context.Context

	TraceBatchQuery(ctx context.Context, conn *Conn, data TraceBatchQueryData)
	TraceBatchEnd(ctx context.Context, conn *Conn, data TraceBatchEndData)
}

BatchTracer traces SendBatch.

type ClusterLoadInfo

type ClusterLoadInfo struct {
	// contains filtered or unexported fields
}

func NewClusterLoadInfo

func NewClusterLoadInfo(ctx context.Context, config *ConnConfig) *ClusterLoadInfo

type CollectableRow

type CollectableRow interface {
	FieldDescriptions() []pgconn.FieldDescription
	Scan(dest ...any) error
	Values() ([]any, error)
	RawValues() [][]byte
}

CollectableRow is the subset of Rows methods that a RowToFunc is allowed to call.

type Conn

type Conn struct {
	// contains filtered or unexported fields
}

Conn is a PostgreSQL connection handle. It is not safe for concurrent usage. Use a connection pool to manage access to multiple database connections from multiple goroutines.

func Connect

func Connect(ctx context.Context, connString string) (*Conn, error)

Connect establishes a connection with a PostgreSQL server with a connection string. See pgconn.Connect for details.

func ConnectConfig

func ConnectConfig(ctx context.Context, connConfig *ConnConfig) (*Conn, error)

ConnectConfig establishes a connection with a PostgreSQL server with a configuration struct. connConfig must have been created by ParseConfig.

func ConnectWithOptions

func ConnectWithOptions(ctx context.Context, connString string, options ParseConfigOptions) (*Conn, error)

ConnectWithOptions behaves exactly like Connect with the addition of options. At the present options is only used to provide a GetSSLPassword function.

func (*Conn) Begin

func (c *Conn) Begin(ctx context.Context) (Tx, error)

Begin starts a transaction. Unlike database/sql, the context only affects the begin command. i.e. there is no auto-rollback on context cancellation.

func (*Conn) BeginTx

func (c *Conn) BeginTx(ctx context.Context, txOptions TxOptions) (Tx, error)

BeginTx starts a transaction with 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.

func (*Conn) Close

func (c *Conn) Close(ctx context.Context) error

Close closes a connection. It is safe to call Close on an already closed connection.

func (*Conn) Config

func (c *Conn) Config() *ConnConfig

Config returns a copy of config that was used to establish this connection.

func (*Conn) CopyFrom

func (c *Conn) CopyFrom(ctx context.Context, tableName Identifier, columnNames []string, rowSrc CopyFromSource) (int64, error)

CopyFrom uses the PostgreSQL copy protocol to perform bulk data insertion. It returns the number of rows copied and an error.

CopyFrom requires all values use the binary format. A pgtype.Type that supports the binary format must be registered for the type of each column. Almost all types implemented by pgx support the binary format.

Even though enum types appear to be strings they still must be registered to use with CopyFrom. This can be done with Conn.LoadType and pgtype.Map.RegisterType.

func (*Conn) Deallocate

func (c *Conn) Deallocate(ctx context.Context, name string) error

Deallocate releases a prepared statement. Calling Deallocate on a non-existent prepared statement will succeed.

func (*Conn) DeallocateAll

func (c *Conn) DeallocateAll(ctx context.Context) error

DeallocateAll releases all previously prepared statements from the server and client, where it also resets the statement and description cache.

func (*Conn) Exec

func (c *Conn) Exec(ctx context.Context, sql string, arguments ...any) (pgconn.CommandTag, error)

Exec executes sql. sql can be either a prepared statement name or an SQL string. arguments should be referenced positionally from the sql string as $1, $2, etc.

func (*Conn) IsClosed

func (c *Conn) IsClosed() bool

IsClosed reports if the connection has been closed.

func (*Conn) LoadType

func (c *Conn) LoadType(ctx context.Context, typeName string) (*pgtype.Type, error)

LoadType inspects the database for typeName and produces a pgtype.Type suitable for registration. typeName must be the name of a type where the underlying type(s) is already understood by pgx. It is for derived types. In particular, typeName must be one of the following:

• An array type name of a type that is already registered. e.g. "_foo" when "foo" is registered.
• A composite type name where all field types are already registered.
• A domain type name where the base type is already registered.
• An enum type name.
• A range type name where the element type is already registered.
• A multirange type name where the element type is already registered.

func (*Conn) PgConn

func (c *Conn) PgConn() *pgconn.PgConn

PgConn returns the underlying *pgconn.PgConn. This is an escape hatch method that allows lower level access to the PostgreSQL connection than pgx exposes.

It is strongly recommended that the connection be idle (no in-progress queries) before the underlying *pgconn.PgConn is used and the connection must be returned to the same state before any *pgx.Conn methods are again used.

func (*Conn) Ping

func (c *Conn) Ping(ctx context.Context) error

Ping delegates to the underlying *pgconn.PgConn.Ping.

func (*Conn) Prepare

func (c *Conn) Prepare(ctx context.Context, name, sql string) (sd *pgconn.StatementDescription, err error)

Prepare creates a prepared statement with name and sql. sql can contain placeholders for bound parameters. These placeholders are referenced positionally as $1, $2, etc. name can be used instead of sql with Query, QueryRow, and Exec to execute the statement. It can also be used with Batch.Queue.

The underlying PostgreSQL identifier for the prepared statement will be name if name != sql or a digest of sql if name == sql.

Prepare is idempotent; i.e. it is safe to call Prepare multiple times with the same name and sql arguments. This allows a code path to Prepare and Query/Exec without concern for if the statement has already been prepared.

func (*Conn) Query

func (c *Conn) Query(ctx context.Context, sql string, args ...any) (Rows, error)

Query sends a query to the server and returns a Rows to read the results. Only errors encountered sending the query and initializing Rows will be returned. Err() on the returned Rows must be checked after the Rows is closed to determine if the query executed successfully.

The returned Rows must be closed before the connection can be used again. It is safe to attempt to read from the returned Rows even if an error is returned. The error will be the available in rows.Err() after rows are closed. It is allowed to ignore the error returned from Query and handle it in Rows.

It is possible for a call of FieldDescriptions on the returned Rows to return nil even if the Query call did not return an error.

It is possible for a query to return one or more rows before encountering an error. In most cases the rows should be collected before processing rather than processed while receiving each row. This avoids the possibility of the application processing rows from a query that the server rejected. The CollectRows function is useful here.

An implementor of QueryRewriter may be passed as the first element of args. It can rewrite the sql and change or replace args. For example, NamedArgs is QueryRewriter that implements named arguments.

For extra control over how the query is executed, the types QueryExecMode, QueryResultFormats, and QueryResultFormatsByOID may be used as the first args to control exactly how the query is executed. This is rarely needed. See the documentation for those types for details.

Example

This example uses Query without using any helpers to read the results. Normally CollectRows, ForEachRow, or another helper function should be used.

ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second)
defer cancel()

conn, err := pgx.Connect(ctx, os.Getenv("PGX_TEST_DATABASE"))
if err != nil {
	fmt.Printf("Unable to establish connection: %v", err)
	return
}

if conn.PgConn().ParameterStatus("crdb_version") != "" {
	// Skip test / example when running on CockroachDB. Since an example can't be skipped fake success instead.
	fmt.Println(`Cheeseburger: $10
Fries: $5
Soft Drink: $3`)
	return
}

// Setup example schema and data.
_, err = conn.Exec(ctx, `
create temporary table products (
	id int primary key generated by default as identity,
	name varchar(100) not null,
	price int not null
);

insert into products (name, price) values
	('Cheeseburger', 10),
	('Double Cheeseburger', 14),
	('Fries', 5),
	('Soft Drink', 3);
`)
if err != nil {
	fmt.Printf("Unable to setup example schema and data: %v", err)
	return
}

rows, err := conn.Query(ctx, "select name, price from products where price < $1 order by price desc", 12)

// It is unnecessary to check err. If an error occurred it will be returned by rows.Err() later. But in rare
// cases it may be useful to detect the error as early as possible.
if err != nil {
	fmt.Printf("Query error: %v", err)
	return
}

// Ensure rows is closed. It is safe to close rows multiple times.
defer rows.Close()

// Iterate through the result set
for rows.Next() {
	var name string
	var price int32

	err = rows.Scan(&name, &price)
	if err != nil {
		fmt.Printf("Scan error: %v", err)
		return
	}

	fmt.Printf("%s: $%d\n", name, price)
}

// rows is closed automatically when rows.Next() returns false so it is not necessary to manually close rows.

// The first error encountered by the original Query call, rows.Next or rows.Scan will be returned here.
if rows.Err() != nil {
	fmt.Printf("rows error: %v", rows.Err())
	return
}

Output:

Cheeseburger: $10
Fries: $5
Soft Drink: $3

func (*Conn) QueryRow

func (c *Conn) QueryRow(ctx context.Context, sql string, args ...any) Row

QueryRow is a convenience wrapper over Query. Any error that occurs while querying is deferred until calling Scan on the returned Row. That Row will error with ErrNoRows if no rows are returned.

func (*Conn) SendBatch

func (c *Conn) SendBatch(ctx context.Context, b *Batch) (br BatchResults)

SendBatch sends all queued queries to the server at once. All queries are run in an implicit transaction unless explicit transaction control statements are executed. The returned BatchResults must be closed before the connection is used again.

Example

ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second)
defer cancel()

conn, err := pgx.Connect(ctx, os.Getenv("PGX_TEST_DATABASE"))
if err != nil {
	fmt.Printf("Unable to establish connection: %v", err)
	return
}

batch := &pgx.Batch{}
batch.Queue("select 1 + 1").QueryRow(func(row pgx.Row) error {
	var n int32
	err := row.Scan(&n)
	if err != nil {
		return err
	}

	fmt.Println(n)

	return err
})

batch.Queue("select 1 + 2").QueryRow(func(row pgx.Row) error {
	var n int32
	err := row.Scan(&n)
	if err != nil {
		return err
	}

	fmt.Println(n)

	return err
})

batch.Queue("select 2 + 3").QueryRow(func(row pgx.Row) error {
	var n int32
	err := row.Scan(&n)
	if err != nil {
		return err
	}

	fmt.Println(n)

	return err
})

err = conn.SendBatch(ctx, batch).Close()
if err != nil {
	fmt.Printf("SendBatch error: %v", err)
	return
}

Output:

2
3
5

func (*Conn) TypeMap

func (c *Conn) TypeMap() *pgtype.Map

TypeMap returns the connection info used for this connection.

func (*Conn) WaitForNotification

func (c *Conn) WaitForNotification(ctx context.Context) (*pgconn.Notification, error)

WaitForNotification waits for a PostgreSQL notification. It wraps the underlying pgconn notification system in a slightly more convenient form.

type ConnConfig

type ConnConfig struct {
	pgconn.Config

	Tracer QueryTracer

	// StatementCacheCapacity is maximum size of the statement cache used when executing a query with "cache_statement"
	// query exec mode.
	StatementCacheCapacity int

	// DescriptionCacheCapacity is the maximum size of the description cache used when executing a query with
	// "cache_describe" query exec mode.
	DescriptionCacheCapacity int

	// DefaultQueryExecMode controls the default mode for executing queries. By default pgx uses the extended protocol
	// and automatically prepares and caches prepared statements. However, this may be incompatible with proxies such as
	// PGBouncer. In this case it may be preferable to use QueryExecModeExec or QueryExecModeSimpleProtocol. The same
	// functionality can be controlled on a per query basis by passing a QueryExecMode as the first query argument.
	DefaultQueryExecMode QueryExecMode
	// contains filtered or unexported fields
}

ConnConfig contains all the options used to establish a connection. It must be created by ParseConfig and then it can be modified. A manually initialized ConnConfig will cause ConnectConfig to panic.

func ParseConfig

func ParseConfig(connString string) (*ConnConfig, error)

func ParseConfigWithOptions

func ParseConfigWithOptions(connString string, options ParseConfigOptions) (*ConnConfig, error)

ParseConfigWithOptions behaves exactly as ParseConfig does with the addition of options. At the present options is only used to provide a GetSSLPassword function.

func (*ConnConfig) ConnString

func (cc *ConnConfig) ConnString() string

ConnString returns the connection string as parsed by pgx.ParseConfig into pgx.ConnConfig.

func (*ConnConfig) Copy

func (cc *ConnConfig) Copy() *ConnConfig

Copy returns a deep copy of the config that is safe to use and modify. The only exception is the tls.Config: according to the tls.Config docs it must not be modified after creation.

type ConnectTracer

type ConnectTracer interface {
	// TraceConnectStart is called at the beginning of Connect and ConnectConfig calls. The returned context is used for
	// the rest of the call and will be passed to TraceConnectEnd.
	TraceConnectStart(ctx context.Context, data TraceConnectStartData) context.Context

	TraceConnectEnd(ctx context.Context, data TraceConnectEndData)
}

ConnectTracer traces Connect and ConnectConfig.

type CopyFromSource

type CopyFromSource interface {
	// Next returns true if there is another row and makes the next row data
	// available to Values(). When there are no more rows available or an error
	// has occurred it returns false.
	Next() bool

	// Values returns the values for the current row.
	Values() ([]any, error)

	// Err returns any error that has been encountered by the CopyFromSource. If
	// this is not nil *Conn.CopyFrom will abort the copy.
	Err() error
}

CopyFromSource is the interface used by *Conn.CopyFrom as the source for copy data.

func CopyFromFunc

func CopyFromFunc(nxtf func() (row []any, err error)) CopyFromSource

CopyFromFunc returns a CopyFromSource interface that relies on nxtf for values. nxtf returns rows until it either signals an 'end of data' by returning row=nil and err=nil, or it returns an error. If nxtf returns an error, the copy is aborted.

func CopyFromRows

func CopyFromRows(rows [][]any) CopyFromSource

CopyFromRows returns a CopyFromSource interface over the provided rows slice making it usable by *Conn.CopyFrom.

func CopyFromSlice

func CopyFromSlice(length int, next func(int) ([]any, error)) CopyFromSource

CopyFromSlice returns a CopyFromSource interface over a dynamic func making it usable by *Conn.CopyFrom.

type CopyFromTracer

type CopyFromTracer interface {
	// TraceCopyFromStart is called at the beginning of CopyFrom calls. The returned context is used for the
	// rest of the call and will be passed to TraceCopyFromEnd.
	TraceCopyFromStart(ctx context.Context, conn *Conn, data TraceCopyFromStartData) context.Context

	TraceCopyFromEnd(ctx context.Context, conn *Conn, data TraceCopyFromEndData)
}

CopyFromTracer traces CopyFrom.

type ExtendedQueryBuilder

type ExtendedQueryBuilder struct {
	ParamValues [][]byte

	ParamFormats  []int16
	ResultFormats []int16
	// contains filtered or unexported fields
}

ExtendedQueryBuilder is used to choose the parameter formats, to format the parameters and to choose the result formats for an extended query.

func (*ExtendedQueryBuilder) Build

func (eqb *ExtendedQueryBuilder) Build(m *pgtype.Map, sd *pgconn.StatementDescription, args []any) error

Build sets ParamValues, ParamFormats, and ResultFormats for use with *PgConn.ExecParams or *PgConn.ExecPrepared. If sd is nil then QueryExecModeExec behavior will be used.

type Identifier

type Identifier []string

Identifier a PostgreSQL identifier or name. Identifiers can be composed of multiple parts such as ["schema", "table"] or ["table", "column"].

func (Identifier) Sanitize

func (ident Identifier) Sanitize() string

Sanitize returns a sanitized string safe for SQL interpolation.

type LargeObject

type LargeObject struct {
	// contains filtered or unexported fields
}

A LargeObject is a large object stored on the server. It is only valid within the transaction that it was initialized in. It uses the context it was initialized with for all operations. It implements these interfaces:

io.Writer
io.Reader
io.Seeker
io.Closer

func (*LargeObject) Close

func (o *LargeObject) Close() error

Close the large object descriptor.

func (*LargeObject) Read

func (o *LargeObject) Read(p []byte) (int, error)

Read reads up to len(p) bytes into p returning the number of bytes read.

func (*LargeObject) Seek

func (o *LargeObject) Seek(offset int64, whence int) (n int64, err error)

Seek moves the current location pointer to the new location specified by offset.

func (*LargeObject) Tell

func (o *LargeObject) Tell() (n int64, err error)

Tell returns the current read or write location of the large object descriptor.

func (*LargeObject) Truncate

func (o *LargeObject) Truncate(size int64) (err error)

Truncate the large object to size.

func (*LargeObject) Write

func (o *LargeObject) Write(p []byte) (int, error)

Write writes p to the large object and returns the number of bytes written and an error if not all of p was written.

type LargeObjectMode

type LargeObjectMode int32

const (
	LargeObjectModeWrite LargeObjectMode = 0x20000
	LargeObjectModeRead  LargeObjectMode = 0x40000
)

type LargeObjects

type LargeObjects struct {
	// contains filtered or unexported fields
}

LargeObjects is a structure used to access the large objects API. It is only valid within the transaction where it was created.

For more details see: http://www.postgresql.org/docs/current/static/largeobjects.html

func (*LargeObjects) Create

func (o *LargeObjects) Create(ctx context.Context, oid uint32) (uint32, error)

Create creates a new large object. If oid is zero, the server assigns an unused OID.

func (*LargeObjects) Open

func (o *LargeObjects) Open(ctx context.Context, oid uint32, mode LargeObjectMode) (*LargeObject, error)

Open opens an existing large object with the given mode. ctx will also be used for all operations on the opened large object.

func (*LargeObjects) Unlink

func (o *LargeObjects) Unlink(ctx context.Context, oid uint32) error

Unlink removes a large object from the database.

type NamedArgs

type NamedArgs map[string]any

NamedArgs can be used as the first argument to a query method. It will replace every '@' named placeholder with a '$' ordinal placeholder and construct the appropriate arguments.

For example, the following two queries are equivalent:

conn.Query(ctx, "select * from widgets where foo = @foo and bar = @bar", pgx.NamedArgs{"foo": 1, "bar": 2})
conn.Query(ctx, "select * from widgets where foo = $1 and bar = $2", 1, 2)

Named placeholders are case sensitive and must start with a letter or underscore. Subsequent characters can be letters, numbers, or underscores.

func (NamedArgs) RewriteQuery

func (na NamedArgs) RewriteQuery(ctx context.Context, conn *Conn, sql string, args []any) (newSQL string, newArgs []any, err error)

RewriteQuery implements the QueryRewriter interface.

type ParseConfigOptions

type ParseConfigOptions struct {
	pgconn.ParseConfigOptions
}

ParseConfigOptions contains options that control how a config is built such as getsslpassword.

type PrepareTracer

type PrepareTracer interface {
	// TracePrepareStart is called at the beginning of Prepare calls. The returned context is used for the
	// rest of the call and will be passed to TracePrepareEnd.
	TracePrepareStart(ctx context.Context, conn *Conn, data TracePrepareStartData) context.Context

	TracePrepareEnd(ctx context.Context, conn *Conn, data TracePrepareEndData)
}

PrepareTracer traces Prepare.

type QueryExecMode

type QueryExecMode int32

const (

	// Automatically prepare and cache statements. This uses the extended protocol. Queries are executed in a single round
	// trip after the statement is cached. This is the default. If the database schema is modified or the search_path is
	// changed after a statement is cached then the first execution of a previously cached query may fail. e.g. If the
	// number of columns returned by a "SELECT *" changes or the type of a column is changed.
	QueryExecModeCacheStatement QueryExecMode

	// Cache statement descriptions (i.e. argument and result types) and assume they do not change. This uses the extended
	// protocol. Queries are executed in a single round trip after the description is cached. If the database schema is
	// modified or the search_path is changed after a statement is cached then the first execution of a previously cached
	// query may fail. e.g. If the number of columns returned by a "SELECT *" changes or the type of a column is changed.
	QueryExecModeCacheDescribe

	// Get the statement description on every execution. This uses the extended protocol. Queries require two round trips
	// to execute. It does not use named prepared statements. But it does use the unnamed prepared statement to get the
	// statement description on the first round trip and then uses it to execute the query on the second round trip. This
	// may cause problems with connection poolers that switch the underlying connection between round trips. It is safe
	// even when the the database schema is modified concurrently.
	QueryExecModeDescribeExec

	// Assume the PostgreSQL query parameter types based on the Go type of the arguments. This uses the extended protocol
	// with text formatted parameters and results. Queries are executed in a single round trip. Type mappings can be
	// registered with pgtype.Map.RegisterDefaultPgType. Queries will be rejected that have arguments that are
	// unregistered or ambiguous. e.g. A map[string]string may have the PostgreSQL type json or hstore. Modes that know
	// the PostgreSQL type can use a map[string]string directly as an argument. This mode cannot.
	QueryExecModeExec

	// Use the simple protocol. Assume the PostgreSQL query parameter types based on the Go type of the arguments.
	// Queries are executed in a single round trip. Type mappings can be registered with
	// pgtype.Map.RegisterDefaultPgType. Queries will be rejected that have arguments that are unregistered or ambiguous.
	// e.g. A map[string]string may have the PostgreSQL type json or hstore. Modes that know the PostgreSQL type can use
	// a map[string]string directly as an argument. This mode cannot.
	//
	// QueryExecModeSimpleProtocol should have the user application visible behavior as QueryExecModeExec with minor
	// exceptions such as behavior when multiple result returning queries are erroneously sent in a single string.
	//
	// QueryExecModeSimpleProtocol uses client side parameter interpolation. All values are quoted and escaped. Prefer
	// QueryExecModeExec over QueryExecModeSimpleProtocol whenever possible. In general QueryExecModeSimpleProtocol
	// should only be used if connecting to a proxy server, connection pool server, or non-PostgreSQL server that does
	// not support the extended protocol.
	QueryExecModeSimpleProtocol
)

func (QueryExecMode) String

func (m QueryExecMode) String() string

type QueryResultFormats

type QueryResultFormats []int16

QueryResultFormats controls the result format (text=0, binary=1) of a query by result column position.

type QueryResultFormatsByOID

type QueryResultFormatsByOID map[uint32]int16

QueryResultFormatsByOID controls the result format (text=0, binary=1) of a query by the result column OID.

type QueryRewriter

type QueryRewriter interface {
	RewriteQuery(ctx context.Context, conn *Conn, sql string, args []any) (newSQL string, newArgs []any, err error)
}

QueryRewriter rewrites a query when used as the first arguments to a query method.

type QueryTracer

type QueryTracer interface {
	// TraceQueryStart is called at the beginning of Query, QueryRow, and Exec calls. The returned context is used for the
	// rest of the call and will be passed to TraceQueryEnd.
	TraceQueryStart(ctx context.Context, conn *Conn, data TraceQueryStartData) context.Context

	TraceQueryEnd(ctx context.Context, conn *Conn, data TraceQueryEndData)
}

QueryTracer traces Query, QueryRow, and Exec.

type QueuedQuery

type QueuedQuery struct {
	SQL       string
	Arguments []any
	// contains filtered or unexported fields
}

QueuedQuery is a query that has been queued for execution via a Batch.

func (*QueuedQuery) Exec

func (qq *QueuedQuery) Exec(fn func(ct pgconn.CommandTag) error)

Exec sets fn to be called when the response to qq is received.

func (*QueuedQuery) Query

func (qq *QueuedQuery) Query(fn func(rows Rows) error)

Query sets fn to be called when the response to qq is received.

func (*QueuedQuery) QueryRow

func (qq *QueuedQuery) QueryRow(fn func(row Row) error)

Query sets fn to be called when the response to qq is received.

type Row

type Row interface {
	// Scan works the same as Rows. with the following exceptions. If no
	// rows were found it returns ErrNoRows. If multiple rows are returned it
	// ignores all but the first.
	Scan(dest ...any) error
}

Row is a convenience wrapper over Rows that is returned by QueryRow.

Row is an interface instead of a struct to allow tests to mock QueryRow. However, adding a method to an interface is technically a breaking change. Because of this the Row interface is partially excluded from semantic version requirements. Methods will not be removed or changed, but new methods may be added.

type RowScanner

type RowScanner interface {
	// ScanRows scans the row.
	ScanRow(rows Rows) error
}

RowScanner scans an entire row at a time into the RowScanner.

type RowToFunc

type RowToFunc[T any] func(row CollectableRow) (T, error)

RowToFunc is a function that scans or otherwise converts row to a T.

type Rows

type Rows interface {
	// Close closes the rows, making the connection ready for use again. It is safe
	// to call Close after rows is already closed.
	Close()

	// Err returns any error that occurred while reading. Err must only be called after the Rows is closed (either by
	// calling Close or by Next returning false). If it is called early it may return nil even if there was an error
	// executing the query.
	Err() error

	// CommandTag returns the command tag from this query. It is only available after Rows is closed.
	CommandTag() pgconn.CommandTag

	// FieldDescriptions returns the field descriptions of the columns. It may return nil. In particular this can occur
	// when there was an error executing the query.
	FieldDescriptions() []pgconn.FieldDescription

	// Next prepares the next row for reading. It returns true if there is another
	// row and false if no more rows are available or a fatal error has occurred.
	// It automatically closes rows when all rows are read.
	//
	// Callers should check rows.Err() after rows.Next() returns false to detect
	// whether result-set reading ended prematurely due to an error. See
	// Conn.Query for details.
	//
	// For simpler error handling, consider using the higher-level pgx v5
	// CollectRows() and ForEachRow() helpers instead.
	Next() bool

	// Scan reads the values from the current row into dest values positionally.
	// dest can include pointers to core types, values implementing the Scanner
	// interface, and nil. nil will skip the value entirely. It is an error to
	// call Scan without first calling Next() and checking that it returned true.
	Scan(dest ...any) error

	// Values returns the decoded row values. As with Scan(), it is an error to
	// call Values without first calling Next() and checking that it returned
	// true.
	Values() ([]any, error)

	// RawValues returns the unparsed bytes of the row values. The returned data is only valid until the next Next
	// call or the Rows is closed.
	RawValues() [][]byte

	// Conn returns the underlying *Conn on which the query was executed. This may return nil if Rows did not come from a
	// *Conn (e.g. if it was created by RowsFromResultReader)
	Conn() *Conn
}

Rows is the result set returned from *Conn.Query. Rows must be closed before the *Conn can be used again. Rows are closed by explicitly calling Close(), calling Next() until it returns false, or when a fatal error occurs.

Once a Rows is closed the only methods that may be called are Close(), Err(), and CommandTag().

Rows is an interface instead of a struct to allow tests to mock Query. However, adding a method to an interface is technically a breaking change. Because of this the Rows interface is partially excluded from semantic version requirements. Methods will not be removed or changed, but new methods may be added.

func RowsFromResultReader

func RowsFromResultReader(typeMap *pgtype.Map, resultReader *pgconn.ResultReader) Rows

RowsFromResultReader returns a Rows that will read from values resultReader and decode with typeMap. It can be used to read from the lower level pgconn interface.

type ScanArgError

type ScanArgError struct {
	ColumnIndex int
	Err         error
}

func (ScanArgError) Error

func (e ScanArgError) Error() string

func (ScanArgError) Unwrap

func (e ScanArgError) Unwrap() error

type TraceBatchEndData

type TraceBatchEndData struct {
	Err error
}

type TraceBatchQueryData

type TraceBatchQueryData struct {
	SQL        string
	Args       []any
	CommandTag pgconn.CommandTag
	Err        error
}

type TraceBatchStartData

type TraceBatchStartData struct {
	Batch *Batch
}

type TraceConnectEndData

type TraceConnectEndData struct {
	Conn *Conn
	Err  error
}

type TraceConnectStartData

type TraceConnectStartData struct {
	ConnConfig *ConnConfig
}

type TraceCopyFromEndData

type TraceCopyFromEndData struct {
	CommandTag pgconn.CommandTag
	Err        error
}

type TraceCopyFromStartData

type TraceCopyFromStartData struct {
	TableName   Identifier
	ColumnNames []string
}

type TracePrepareEndData

type TracePrepareEndData struct {
	AlreadyPrepared bool
	Err             error
}

type TracePrepareStartData

type TracePrepareStartData struct {
	Name string
	SQL  string
}

type TraceQueryEndData

type TraceQueryEndData struct {
	CommandTag pgconn.CommandTag
	Err        error
}

type TraceQueryStartData

type TraceQueryStartData struct {
	SQL  string
	Args []any
}

type Tx

type Tx interface {
	// Begin starts a pseudo nested transaction.
	Begin(ctx context.Context) (Tx, error)

	// Commit commits the transaction if this is a real transaction or releases the savepoint if this is a pseudo nested
	// transaction. Commit will return an error where errors.Is(ErrTxClosed) is true 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 an error where errors.Is(ErrTxCommitRollback) is true will be returned.
	Commit(ctx context.Context) error

	// Rollback rolls back the transaction if this is a real transaction or rolls back to the savepoint if this is a
	// pseudo nested transaction. Rollback will return an error where errors.Is(ErrTxClosed) is true if the Tx is already
	// closed, but is otherwise safe to call multiple times. Hence, a defer tx.Rollback() is safe even if tx.Commit() will
	// be called first in a non-error condition. Any other failure of a real transaction will result in the connection
	// being closed.
	Rollback(ctx context.Context) error

	CopyFrom(ctx context.Context, tableName Identifier, columnNames []string, rowSrc CopyFromSource) (int64, error)
	SendBatch(ctx context.Context, b *Batch) BatchResults
	LargeObjects() LargeObjects

	Prepare(ctx context.Context, name, sql string) (*pgconn.StatementDescription, error)

	Exec(ctx context.Context, sql string, arguments ...any) (commandTag pgconn.CommandTag, err error)
	Query(ctx context.Context, sql string, args ...any) (Rows, error)
	QueryRow(ctx context.Context, sql string, args ...any) Row

	// Conn returns the underlying *Conn that on which this transaction is executing.
	Conn() *Conn
}

Tx represents a database transaction.

Tx is an interface instead of a struct to enable connection pools to be implemented without relying on internal pgx state, to support pseudo-nested transactions with savepoints, and to allow tests to mock transactions. However, adding a method to an interface is technically a breaking change. If new methods are added to Conn it may be desirable to add them to Tx as well. Because of this the Tx interface is partially excluded from semantic version requirements. Methods will not be removed or changed, but new methods may be added.

type TxAccessMode

type TxAccessMode string

TxAccessMode is the transaction access mode (read write or read only)

const (
	ReadWrite TxAccessMode = "read write"
	ReadOnly  TxAccessMode = "read only"
)

Transaction access modes

type TxDeferrableMode

type TxDeferrableMode string

TxDeferrableMode is the transaction deferrable mode (deferrable or not deferrable)

const (
	Deferrable    TxDeferrableMode = "deferrable"
	NotDeferrable TxDeferrableMode = "not deferrable"
)

Transaction deferrable modes

type TxIsoLevel

type TxIsoLevel string

TxIsoLevel is the transaction isolation level (serializable, repeatable read, read committed or read uncommitted)

const (
	Serializable    TxIsoLevel = "serializable"
	RepeatableRead  TxIsoLevel = "repeatable read"
	ReadCommitted   TxIsoLevel = "read committed"
	ReadUncommitted TxIsoLevel = "read uncommitted"
)

Transaction isolation levels

type TxOptions

type TxOptions struct {
	IsoLevel       TxIsoLevel
	AccessMode     TxAccessMode
	DeferrableMode TxDeferrableMode

	// BeginQuery is the SQL query that will be executed to begin the transaction. This allows using non-standard syntax
	// such as BEGIN PRIORITY HIGH with CockroachDB. If set this will override the other settings.
	BeginQuery string
}

TxOptions are transaction modes within a transaction block