harmonydb

Documentation

Overview

HarmonyDB provides database abstractions over SP-wide Postgres-compatible instance(s).

Features

Rolling to secondary database servers on connection failure
Convenience features for Go + SQL
Prevention of SQL injection vulnerabilities
Monitors behavior via Prometheus stats and logging of errors.

Usage

Processes should use New() to instantiate a *DB and keep it. Consumers can use this *DB concurrently. Creating and changing tables & views should happen in ./sql/ folder. Name the file "today's date" in the format: YYYYMMDD.sql (ex: 20231231.sql for the year's last day)

a. CREATE TABLE should NOT have a schema:
	GOOD: CREATE TABLE foo ();
	BAD:  CREATE TABLE me.foo ();
b. Schema is managed for you. It provides isolation for integraton tests & multi-use.
c. Git Merges: All run once, so old-after-new is OK when there are no deps.
d. NEVER change shipped sql files. Have later files make corrections.
e. Anything not ran will be ran, so an older date making it to master is OK.

Write SQL with context, raw strings, and args:

name := "Alice"
var ID int
err := QueryRow(ctx, "SELECT id from people where first_name=?", name).Scan(&ID)
fmt.Println(ID)

Note: Scan() is column-oriented, while Select() & StructScan() is field name/tag oriented.

Index

• Constants
• Variables
• func IsErrSerialization(err error) bool
• func IsErrUniqueContraint(err error) bool
• func NewFromConfigWithITestID(cfg config.HarmonyDB) func(id ITestID) (*DB, error)
• type DB
  • func New(hosts []string, username, password, database, port string, itestID ITestID) (*DB, error)
  • func NewFromConfig(cfg config.HarmonyDB) (*DB, error)
  • func (db *DB) BeginTransaction(ctx context.Context, f func(*Tx) (commit bool, err error), ...) (didCommit bool, retErr error)
  • func (db *DB) Exec(ctx context.Context, sql rawStringOnly, arguments ...any) (count int, err error)
  • func (db *DB) GetRoutableIP() (string, error)
  • func (db *DB) ITestDeleteAll()
  • func (db *DB) Query(ctx context.Context, sql rawStringOnly, arguments ...any) (*Query, error)
  • func (db *DB) QueryRow(ctx context.Context, sql rawStringOnly, arguments ...any) Row
  • func (db *DB) Select(ctx context.Context, sliceOfStructPtr any, sql rawStringOnly, arguments ...any) error
• type ITestID
  • func ITestNewID() ITestID
• type Qry
• type Query
  • func (q *Query) StructScan(s any) error
• type Row
• type TransactionOption
  • func OptionRetry() TransactionOption
  • func OptionSerialRetryTime(d time.Duration) TransactionOption
• type TransactionOptions
• type Tx
  • func (t *Tx) Exec(sql rawStringOnly, arguments ...any) (count int, err error)
  • func (t *Tx) Query(sql rawStringOnly, arguments ...any) (*Query, error)
  • func (t *Tx) QueryRow(sql rawStringOnly, arguments ...any) Row
  • func (t *Tx) Select(sliceOfStructPtr any, sql rawStringOnly, arguments ...any) error

Constants

const SQL_START = ctxkey("sqlStart")

const SQL_STRING = ctxkey("sqlString")

Variables

var DBMeasures = struct {
	Hits            *stats.Int64Measure
	TotalWait       *stats.Int64Measure
	Waits           prometheus.Histogram
	OpenConnections *stats.Int64Measure
	Errors          *stats.Int64Measure
	WhichHost       prometheus.Histogram
}{
	Hits:      stats.Int64(pre+"hits", "Total number of uses.", stats.UnitDimensionless),
	TotalWait: stats.Int64(pre+"total_wait", "Total delay. A numerator over hits to get average wait.", stats.UnitMilliseconds),
	Waits: prometheus.NewHistogram(prometheus.HistogramOpts{
		Name:    pre + "waits",
		Buckets: waitsBuckets,
		Help:    "The histogram of waits for query completions.",
	}),
	OpenConnections: stats.Int64(pre+"open_connections", "Total connection count.", stats.UnitDimensionless),
	Errors:          stats.Int64(pre+"errors", "Total error count.", stats.UnitDimensionless),
	WhichHost: prometheus.NewHistogram(prometheus.HistogramOpts{
		Name:    pre + "which_host",
		Buckets: whichHostBuckets,
		Help:    "The index of the hostname being used",
	}),
}

DBMeasures groups all db metrics.

Functions

func IsErrSerialization

func IsErrSerialization(err error) bool

func IsErrUniqueContraint

func IsErrUniqueContraint(err error) bool

func NewFromConfigWithITestID

func NewFromConfigWithITestID(cfg config.HarmonyDB) func(id ITestID) (*DB, error)

Types

type DB

type DB struct {
	BTFPOnce sync.Once
	BTFP     atomic.Uintptr // BeginTransactionFramePointer
	// contains filtered or unexported fields
}

func New

func New(hosts []string, username, password, database, port string, itestID ITestID) (*DB, error)

New is to be called once per binary to establish the pool. log() is for errors. It returns an upgraded database's connection. This entry point serves both production and integration tests, so it's more DI.

func NewFromConfig

func NewFromConfig(cfg config.HarmonyDB) (*DB, error)

NewFromConfig is a convenience function. In usage:

db, err := NewFromConfig(config.HarmonyDB)  // in binary init

func (*DB) BeginTransaction

func (db *DB) BeginTransaction(ctx context.Context, f func(*Tx) (commit bool, err error), opt ...TransactionOption) (didCommit bool, retErr error)

BeginTransaction is how you can access transactions using this library. The entire transaction happens in the function passed in. The return must be true or a rollback will occur. Be sure to test the error for IsErrSerialization() if you want to retry

when there is a DB serialization error.

func (*DB) Exec

func (db *DB) Exec(ctx context.Context, sql rawStringOnly, arguments ...any) (count int, err error)

Exec executes changes (INSERT, DELETE, or UPDATE). Note, for CREATE & DROP please keep these permanent and express them in the ./sql/ files (next number).

func (*DB) GetRoutableIP

func (db *DB) GetRoutableIP() (string, error)

func (*DB) ITestDeleteAll

func (db *DB) ITestDeleteAll()

ITestDeleteAll will delete everything created for "this" integration test. This must be called at the end of each integration test.

func (*DB) Query

func (db *DB) Query(ctx context.Context, sql rawStringOnly, arguments ...any) (*Query, error)

Query allows iterating returned values to save memory consumption with the downside of needing to `defer q.Close()`. For a simpler interface, try Select() Next() must be called to advance the row cursor, including the first time: Ex: q, err := db.Query(ctx, "SELECT id, name FROM users") handleError(err) defer q.Close()

for q.Next() {
	  var id int
   var name string
   handleError(q.Scan(&id, &name))
   fmt.Println(id, name)
}

func (*DB) QueryRow

func (db *DB) QueryRow(ctx context.Context, sql rawStringOnly, arguments ...any) Row

QueryRow gets 1 row using column order matching. This is a timesaver for the special case of wanting the first row returned only. EX:

var name, pet string
var ID = 123
err := db.QueryRow(ctx, "SELECT name, pet FROM users WHERE ID=?", ID).Scan(&name, &pet)

func (*DB) Select

func (db *DB) Select(ctx context.Context, sliceOfStructPtr any, sql rawStringOnly, arguments ...any) error

Select multiple rows into a slice using name matching Ex:

type user struct {
	Name string
	ID int
	Number string `db:"tel_no"`
}

var users []user
pet := "cat"
err := db.Select(ctx, &users, "SELECT name, id, tel_no FROM customers WHERE pet=?", pet)

type ITestID

type ITestID string

func ITestNewID

func ITestNewID() ITestID

ITestNewID see ITestWithID doc

type Qry

type Qry interface {
	Next() bool
	Err() error
	Close()
	Scan(...any) error
	Values() ([]any, error)
}

type Query

type Query struct {
	Qry
}

Query offers Next/Err/Close/Scan/Values

func (*Query) StructScan

func (q *Query) StructScan(s any) error

StructScan allows scanning a single row into a struct. This improves efficiency of processing large result sets by avoiding the need to allocate a slice of structs.

type Row

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

type TransactionOption

type TransactionOption func(*TransactionOptions)

func OptionRetry

func OptionRetry() TransactionOption

func OptionSerialRetryTime

func OptionSerialRetryTime(d time.Duration) TransactionOption

type TransactionOptions

type TransactionOptions struct {
	RetrySerializationError            bool
	InitialSerializationErrorRetryWait time.Duration
}

type Tx

type Tx struct {
	pgx.Tx
	// contains filtered or unexported fields
}

func (*Tx) Exec

func (t *Tx) Exec(sql rawStringOnly, arguments ...any) (count int, err error)

Exec in a transaction.

func (*Tx) Query

func (t *Tx) Query(sql rawStringOnly, arguments ...any) (*Query, error)

Query in a transaction.

func (*Tx) QueryRow

func (t *Tx) QueryRow(sql rawStringOnly, arguments ...any) Row

QueryRow in a transaction.

func (*Tx) Select

func (t *Tx) Select(sliceOfStructPtr any, sql rawStringOnly, arguments ...any) error

Select in a transaction.