pgsql

package
v0.8.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 26, 2026 License: Apache-2.0 Imports: 18 Imported by: 0

Documentation

Overview

Package pgsql implements hopper's driver operations for PostgreSQL over a small transport interface, so that the pgx driver (hopperpgx) and the database/sql driver (hoppersql) share one implementation of the SQL and the logic around it.

Everything crosses the transport boundary as portable values: parameters are strings, numbers, booleans, times, plain slices of them and pointers to them for NULL (JSON goes as text, and every statement casts its parameters, so a transport that cannot encode slices renders them with ArrayLiteral), and scan targets are the standard library's types plus a few sql.Scanner implementations here. A transport therefore needs no type registration.

Index

Constants

View Source 
const ClaimMax = 512

ClaimMax is the most jobs one claim statement returns.

View Source 
const HistoryIndexCancelKey int32 = 0x686f7063 // "hopc"

HistoryIndexCancelKey is the pg_advisory_xact_lock key, combined with the schema, that a migration holds from before it cancels the leader's index builds until it commits. A build that fails with 57014, which a statement timeout and a cancel share, was canceled by a migration if the key is held: the message would tell them apart only in the server's language.

View Source 
const HistoryIndexLockKey int32 = 0x686f7069 // "hopi"

HistoryIndexLockKey is the pg_advisory_xact_lock key held while a history partition is indexed, combined with the schema like HistoryLockKey. Two concurrent CREATE INDEX CONCURRENTLY on one table can deadlock, so overlapping leaders index one at a time; the one that does not get the lock leaves the work to the other.

A build also holds the key combined with the schema and its outcome table: exclusively on the DEFAULT partition, in share mode elsewhere. Partition creation under the table tries it in share mode without waiting, and leaves the partition to its next pass while the DEFAULT partition's build runs: its attach scans the DEFAULT partition, which the build holds, while the build waits for the attach's transaction to end. A build elsewhere does not conflict with the attach. Retention tries the table's key as well, and keeps the partitions past it while a build holds it: a drop locks the DEFAULT partition too, and a drop of the partition being built would hold the table locked while it waits for the build. It tries the key in share mode for a partition that has every index, which no build runs on, and exclusively for one that lacks some. A build waits for the table's key, which creation and retention hold only briefly.

The CREATE INDEX CONCURRENTLY cannot run in a transaction, so the locks are held by a transaction of their own, which stays open and idle on another connection for the build. A session-level lock would not do: behind a transaction pooler, the lock, the build and the unlock can each run on a different server connection, and the lock would outlive the build.

View Source 
const HistoryLockKey int32 = 0x686f7068 // "hoph"

HistoryLockKey is the pg_advisory_xact_lock key that serializes changes to the history partitions and their indexes: partition creation between overlapping leaders, and index attaches. It is combined with the schema, so installations sharing a database do not queue behind each other.

View Source 
const MigrationLockKey int64 = 0x686f707065725f6d // "hopper_m"

MigrationLockKey is the pg_advisory_lock key held while migrating.

View Source 
const NotifySQL = `SELECT pg_notify($1, p) FROM unnest($2::text[]) AS p`

NotifySQL sends one notification per payload. Postgres delivers them on commit and collapses identical ones within a transaction.

Variables

View Source 
var ErrNoRows = errors.New("pgsql: no rows")

ErrNoRows is returned by Row.Scan when the query matched nothing. Transports translate their own sentinel to it.

Functions

func ArrayLiteral 

func ArrayLiteral(v any) (literal string, ok bool)

ArrayLiteral renders a slice parameter as a Postgres array literal, for a transport that cannot encode slices itself. ok is false for any other value.

func JobColumns 

func JobColumns(prefix string) string

JobColumns lists hopper_jobs columns in scan order. prefix qualifies them with a table alias.

func MigrationApply 

func MigrationApply(ctx context.Context, conn Conn, schema *Schema, version int, script string, up bool) (err error)

MigrationApply runs a migration script and records (up) or removes (down) its version, atomically. The script may hold several statements, so the transport must run it as a simple query (no parameters).

func MigrationVersions 

func MigrationVersions(ctx context.Context, conn Conn, schema *Schema) ([]int, error)

MigrationVersions returns the applied schema versions, or nil if the schema has never been installed.

func TagsJSON added in v0.8.0 

func TagsJSON(tags []string) []byte

TagsJSON encodes a job's tags for its tags column: a JSON array, or nil for none, which is stored as NULL.

Types

type Conn 

type Conn interface {
	// Exec runs a statement and returns the rows affected.
	Exec(ctx context.Context, sql string, args ...any) (int64, error)
	Query(ctx context.Context, sql string, args ...any) (Rows, error)
	QueryRow(ctx context.Context, sql string, args ...any) Row
	// QueryExec runs a query followed by a statement, pipelined in one
	// round trip where the transport can, and inside one transaction. It
	// returns the query's rows.
	QueryExec(ctx context.Context, query string, queryArgs []any, exec string, execArgs []any) (Rows, error)
	// Begin starts a transaction, or a savepoint inside one.
	Begin(ctx context.Context) (Tx, error)
}

Conn runs statements: a pool, a dedicated connection or a transaction.

type Executor 

type Executor struct {
	Conn   Conn
	Schema *Schema
	InTx   bool
}

Executor implements driver.Executor over a Conn. InTx says the Conn is the caller's transaction, which rules out operations that must not run inside one.

func (*Executor) AnalyzeForTest 

func (e *Executor) AnalyzeForTest(ctx context.Context) error

AnalyzeForTest analyzes the live table as it is, for the conformance suite to give the planner statistics of a known state.

func (*Executor) BatchGet 

func (e *Executor) BatchGet(ctx context.Context, id driver.JobID) (*driver.BatchRow, error)

BatchGet implements driver.Executor.

func (*Executor) BatchInsert 

func (e *Executor) BatchInsert(ctx context.Context, params driver.BatchInsertParams) (driver.JobID, error)

BatchInsert implements driver.Executor.

func (*Executor) BatchSettle added in v0.7.0 

func (e *Executor) BatchSettle(ctx context.Context, params driver.BatchSettleParams) (driver.BatchSettleResult, error)

BatchSettle implements driver.Executor.

func (*Executor) ClientDelete 

func (e *Executor) ClientDelete(ctx context.Context, clientID int64) error

ClientDelete implements driver.Executor.

func (*Executor) ClientList 

func (e *Executor) ClientList(ctx context.Context) ([]*driver.ClientRow, error)

ClientList implements driver.Executor.

func (*Executor) ClientPruneExpired 

func (e *Executor) ClientPruneExpired(ctx context.Context) (int64, error)

ClientPruneExpired implements driver.Executor.

func (*Executor) ClientRegister 

func (e *Executor) ClientRegister(ctx context.Context, params driver.ClientRegisterParams) (int64, error)

ClientRegister implements driver.Executor.

func (*Executor) ClientRenew 

func (e *Executor) ClientRenew(ctx context.Context, params driver.ClientRenewParams) (driver.ClientRenewResult, error)

ClientRenew implements driver.Executor.

func (*Executor) HistoryIndex added in v0.7.0 

func (e *Executor) HistoryIndex(ctx context.Context, params driver.HistoryIndexParams) (driver.HistoryIndexResult, error)

HistoryIndex implements driver.Executor. It builds each history index of every partition that has none attached yet, concurrently so that finalizes keep moving rows into history, and attaches it. Partitions created since an index was declared get theirs when they are attached, so this finds work only after an upgrade, once per partition and index. An index a canceled build left invalid is dropped and built again. A build canceled by the session's statement timeout or by an operator is reported, and the pass goes on to the next partition; a migration's cancel ends the pass.

func (*Executor) HistoryMaintain 

func (e *Executor) HistoryMaintain(ctx context.Context, params driver.HistoryMaintainParams) (driver.HistoryMaintainResult, error)

HistoryMaintain implements driver.Executor. It keeps time partitions ready for the periods ahead and drops the ones past retention. Only future periods are created, so no existing row ever has to move: rows finalized before their period's partition existed (the first period after installation, or after a long leader outage) stay in the DEFAULT partition and are pruned by retention individually.

func (*Executor) JobAge 

func (e *Executor) JobAge(ctx context.Context, queue string, after time.Duration) (int64, error)

JobAge implements driver.Executor.

func (*Executor) JobCancel 

func (e *Executor) JobCancel(ctx context.Context, id driver.JobID) (*driver.JobRow, error)

JobCancel implements driver.Executor.

func (*Executor) JobClaim 

func (e *Executor) JobClaim(ctx context.Context, params driver.JobClaimParams) (driver.JobClaimResult, error)

JobClaim implements driver.Executor.

func (*Executor) JobDelete added in v0.8.0 

func (e *Executor) JobDelete(ctx context.Context, id driver.JobID) (*driver.JobRow, error)

JobDelete implements driver.Executor. A JobRetry that is moving the job from history back to the live table makes the history delete wait for it and then find nothing, so before the job is called unknown the live table, where the retry put it, is read once more.

func (*Executor) JobDiscardExpired 

func (e *Executor) JobDiscardExpired(ctx context.Context, limit int) ([]*driver.JobRow, error)

JobDiscardExpired implements driver.Executor.

func (*Executor) JobFinalizeMany 

func (e *Executor) JobFinalizeMany(ctx context.Context, params driver.JobFinalizeParams) ([]driver.JobID, error)

JobFinalizeMany implements driver.Executor.

func (*Executor) JobGet 

func (e *Executor) JobGet(ctx context.Context, id driver.JobID) (*driver.JobRow, error)

JobGet implements driver.Executor.

func (*Executor) JobInsertCopy 

func (e *Executor) JobInsertCopy(context.Context, []driver.JobInsertParams, driver.JobInsertOpts) ([]driver.JobInsertResult, error)

JobInsertCopy implements driver.Executor for transports without COPY.

func (*Executor) JobInsertMany 

func (e *Executor) JobInsertMany(ctx context.Context, params []driver.JobInsertParams, opts driver.JobInsertOpts) ([]driver.JobInsertResult, error)

JobInsertMany implements driver.Executor.

func (*Executor) JobList 

func (e *Executor) JobList(ctx context.Context, params driver.JobListParams) ([]*driver.JobRow, error)

JobList implements driver.Executor. The ordering key, IDs, priorities and metadata are matched only when set, each as a plain condition, so that the indexes on them serve it: the ordering-key indexes and the primary keys. Tags are found through the tags tables (tagJobsSQL, listTagsHistory).

func (*Executor) JobLogs added in v0.8.0 

func (e *Executor) JobLogs(ctx context.Context, id driver.JobID) ([]driver.AttemptLog, error)

JobLogs implements driver.Executor.

func (*Executor) JobRecordProgress added in v0.8.0 

func (e *Executor) JobRecordProgress(ctx context.Context, params driver.JobRecordProgressParams) (*driver.Progress, error)

JobRecordProgress implements driver.Executor.

func (*Executor) JobRescueCandidates 

func (e *Executor) JobRescueCandidates(ctx context.Context, params driver.JobRescueParams) ([]driver.RescueCandidate, error)

JobRescueCandidates implements driver.Executor.

func (*Executor) JobRetry 

func (e *Executor) JobRetry(ctx context.Context, params driver.JobRetryParams) (*driver.JobRow, error)

JobRetry implements driver.Executor.

func (*Executor) JobRetryOrderingKey added in v0.7.0 

func (e *Executor) JobRetryOrderingKey(ctx context.Context, params driver.JobRetryOrderingKeyParams) (int, error)

JobRetryOrderingKey implements driver.Executor. The mark and the wake are separate statements, in that order, so the wake reads a snapshot taken after the mark: a job that ends its attempt before the mark reaches it is waiting by then and is woken, and one that ends after sees the mark.

func (*Executor) JobsMaintain 

func (e *Executor) JobsMaintain(ctx context.Context) (driver.JobsMaintainResult, error)

JobsMaintain implements driver.Executor. A vacuum of the live table reclaims the index entries every claim would otherwise walk past. What matters is their number, not their share of the table: a hundred thousand dead entries are about five hundred index pages, well under a millisecond per claim, so the vacuum runs once that many have accumulated, and a short burst never triggers one.

ANALYZE keeps two things current that the planner needs: the row count, which a plan cached while the table was empty gets wrong by orders of magnitude, and the mix of states, from which the planner costs scans of the partial indexes by state. (The finalize does not depend on it: it compares the running state as text; see jobFinalizeHeadSQL.) So it runs at once when the planner's row count is ten times off, and otherwise once a tenth of the table has changed but no more than every thirty seconds, since an analyze invalidates every connection's cached plans and a CPU-starved server feels the replanning. The sample is kept small; nothing here needs fine histograms. Both skip, rather than wait, when autovacuum holds the lock, since that pass does the same work.

The live tags table churns with tagged jobs, and a listing by tag walks its index, so it is kept the same way, except that its analyze keeps the server's statistics target. So is the dependency table, which churns with workflows, except that churn re-analyzes it at most every analyzeDepsEvery: until it has statistics, the planner takes each lookup of a job's edges for a two-hundredth of the table, and scans the table instead, in finalizes and in the sweep that promotes dependents (see finalizeTailSQL). The counters of all three are read in one statement, each table's in a LATERAL subquery that OFFSET 0 keeps from being flattened: joined, the view's GROUP BY kept the table's OID from reaching pg_class, so every pass summed the counters of every relation in the database. Each table is kept whether or not another's maintenance fails, such as a vacuum of a large hopper_jobs that runs past the role's statement_timeout on every pass.

func (*Executor) LeaderAttempt 

func (e *Executor) LeaderAttempt(ctx context.Context, params driver.LeaderParams) (bool, error)

LeaderAttempt implements driver.Executor.

func (*Executor) LeaderResign 

func (e *Executor) LeaderResign(ctx context.Context, clientID int64) error

LeaderResign implements driver.Executor.

func (*Executor) MessagePublish 

func (e *Executor) MessagePublish(ctx context.Context, p driver.MessagePublishParams) ([]driver.JobInsertResult, error)

MessagePublish implements driver.Executor.

func (*Executor) Notify 

func (e *Executor) Notify(ctx context.Context, channel string, payloads []string) error

Notify implements driver.Executor.

func (*Executor) Now 

func (e *Executor) Now(ctx context.Context) (time.Time, error)

Now implements driver.Executor.

func (*Executor) PeriodicInsert 

func (e *Executor) PeriodicInsert(ctx context.Context, params driver.PeriodicInsertParams) (*driver.JobRow, bool, error)

PeriodicInsert implements driver.Executor.

func (*Executor) PeriodicLastSlots 

func (e *Executor) PeriodicLastSlots(ctx context.Context) (map[string]time.Time, error)

PeriodicLastSlots implements driver.Executor.

func (*Executor) QueueEnsure 

func (e *Executor) QueueEnsure(ctx context.Context, names []string) error

QueueEnsure implements driver.Executor.

func (*Executor) QueueList 

func (e *Executor) QueueList(ctx context.Context) ([]*driver.QueueRow, error)

QueueList implements driver.Executor.

func (*Executor) QueuePause 

func (e *Executor) QueuePause(ctx context.Context, name string) error

QueuePause implements driver.Executor.

func (*Executor) QueueResume 

func (e *Executor) QueueResume(ctx context.Context, name string) error

QueueResume implements driver.Executor.

func (*Executor) QueueSetLimits 

func (e *Executor) QueueSetLimits(ctx context.Context, l driver.QueueLimits) error

QueueSetLimits implements driver.Executor.

func (*Executor) SchemaVersion added in v0.6.0 

func (e *Executor) SchemaVersion(ctx context.Context) (int, error)

SchemaVersion implements driver.Executor.

func (*Executor) Stats 

func (e *Executor) Stats(ctx context.Context) (*driver.Stats, error)

Stats implements driver.Executor.

func (*Executor) StreamAppend 

func (e *Executor) StreamAppend(ctx context.Context, params driver.StreamAppendParams) (*driver.StreamEvent, error)

StreamAppend implements driver.Executor.

func (*Executor) StreamConsumerList 

func (e *Executor) StreamConsumerList(ctx context.Context) ([]*driver.StreamConsumerRow, error)

StreamConsumerList implements driver.Executor.

func (*Executor) StreamConsumerSeek 

func (e *Executor) StreamConsumerSeek(ctx context.Context, name string, params driver.StreamSeekParams) error

StreamConsumerSeek implements driver.Executor. Seeking to a position or a time resumes the log from a transaction ID onwards: the snapshot "x:x:" sees every transaction below x, and the cursor skips the events of x itself up to the position.

func (*Executor) StreamConsumerUpsert 

func (e *Executor) StreamConsumerUpsert(ctx context.Context, consumers []driver.StreamConsumerRow) error

StreamConsumerUpsert implements driver.Executor.

func (*Executor) StreamPump 

func (e *Executor) StreamPump(ctx context.Context, name string, limit int) (driver.StreamPumpResult, error)

StreamPump implements driver.Executor.

func (*Executor) StreamRead 

func (e *Executor) StreamRead(ctx context.Context, params driver.StreamReadParams) ([]*driver.StreamEvent, error)

StreamRead implements driver.Executor.

func (*Executor) SubscriptionList 

func (e *Executor) SubscriptionList(ctx context.Context) ([]*driver.SubscriptionRow, error)

SubscriptionList implements driver.Executor.

func (*Executor) SubscriptionUpsert 

func (e *Executor) SubscriptionUpsert(ctx context.Context, subs []driver.SubscriptionRow) error

SubscriptionUpsert implements driver.Executor.

func (*Executor) WorkflowGet 

func (e *Executor) WorkflowGet(ctx context.Context, id driver.JobID) (*driver.WorkflowRow, error)

WorkflowGet implements driver.Executor.

func (*Executor) WorkflowInsert 

func (e *Executor) WorkflowInsert(ctx context.Context, params driver.WorkflowInsertParams) (*driver.WorkflowInsertResult, error)

WorkflowInsert implements driver.Executor.

func (*Executor) WorkflowPromoteReady added in v0.7.0 

func (e *Executor) WorkflowPromoteReady(ctx context.Context, params driver.WorkflowPromoteParams) (driver.WorkflowPromoteResult, error)

WorkflowPromoteReady implements driver.Executor.

type Row 

type Row interface {
	Scan(dest ...any) error
}

Row is a single-row result. Scan returns ErrNoRows when there is none.

type Rows 

type Rows interface {
	Next() bool
	Scan(dest ...any) error
	Err() error
	Close()
}

Rows is a result set. Close must be called; Err reports iteration errors.

type Schema added in v0.3.0 

type Schema struct {
	Name string
	// contains filtered or unexported fields
}

Schema binds SQL templates to one namespace. Cached statements are shared by all executors and transactions of a driver.

func NewSchema added in v0.3.0 

func NewSchema(name string) *Schema

NewSchema resolves the default namespace.

func (*Schema) SQL added in v0.3.0 

func (s *Schema) SQL(query string) string

SQL expands explicit namespace placeholders in internal SQL templates.

func (*Schema) Table added in v0.3.0 

func (s *Schema) Table(name string) string

Table returns a safely quoted, schema-qualified object name.

func (*Schema) Wrap added in v0.3.0 

func (s *Schema) Wrap(c Conn) Conn

Wrap qualifies templates without changing the underlying connection.

type Tx 

type Tx interface {
	Conn
	Commit(ctx context.Context) error
	Rollback(ctx context.Context) error
}

Tx is a transaction (or savepoint) on a Conn.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.