pgsql

package
v0.1.1 Latest Latest
Warning

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

 Go to latest Published: Sep 29, 2026 License: Apache-2.0 Imports: 10 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 HistoryLockKey int32 = 0x686f7068 // "hoph"

HistoryLockKey is the pg_advisory_xact_lock key that serializes partition creation between overlapping leaders. 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, 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) ([]int, error)

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

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
	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) 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) 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) 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.

func (*Executor) JobRescueCandidates 

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

JobRescueCandidates implements driver.Executor.

func (*Executor) JobRetry 

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

JobRetry implements driver.Executor.

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, which decides how the finalize joins its parameter list to the running rows (with a stale mix that says almost nothing is running, it probes the list once per running row). 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.

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) 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.

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 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.