postgresadapter

func Distinct ¶

func Distinct(ctx context.Context, database DBTX, specification *query.ResourceSQL, listQuery *query.ListQuery, field query.Field) ([]livewire.DistinctValue, error)

Distinct is the distinct read of D15: the values of field under the list's WHERE with their counts, most frequent first, at most listQuery.Limit of them (the resource's DistinctMaximum, which the engine puts there). It runs in the engine's transaction inside an engine unit of work, on database outside one. Field must be one of the resource's filterable scalar fields, else it is a *query.Error.

func HistoryPage ¶

func HistoryPage[Item any](ctx context.Context, database DBTX, table string, listQuery *query.ListQuery, decode func(row []byte) (*Item, error)) (*livewire.Page[livewire.Version[Item]], error)

HistoryPage reads one page of an item's versions from table's `<table>_history` (HistorySQL) as a resource's History loader answers it (D34, D56): listQuery is the engine's history query (HistoryFields, scoped to the item's id), over the versions of the id's current life only, so an item re-created under a deleted id never shows the deleted one's versions; it is read with its page query and its capped count on one snapshot like ListPage, in the engine's transaction inside an engine unit of work. decode maps a version's stored row, the jsonb of the table's columns, to the item's JSON shape. The page has no cursor: a history pages by offset.

History: func(ctx context.Context, listQuery *query.ListQuery) (*livewire.Page[livewire.Version[part.Item]], error) {
	return postgresadapter.HistoryPage(ctx, r.pool, "parts", listQuery, decodePartRow)
},

func HistorySQL ¶

func HistorySQL(table string) (string, error)

HistorySQL creates `<table>_history` and the trigger that appends a version on every insert, update and delete of table (D34): the row as jsonb in data, the version (1, 2, ... per id), the operation, valid_from and valid_to (NULL for the current version; a delete is a closed version of the last row), and the actor, real actor, mutation and request id the Transactor set in livewire.actor (D31; NULL outside an engine transaction). An update that changes nothing appends nothing, a TRUNCATE appends nothing either; an update that changes the id closes the old id's version and starts the new id's history. Table needs a uuid column id. For a natively partitioned table the history is the table's, whichever partition the row lives in; the id must be unique across the partitions, which PostgreSQL cannot enforce there and livewire assumes everywhere (Change.ID, item routes). An update that moves a row to another partition runs as a delete and an insert, so it appends a delete and an insert version, not an update. The trigger function livewire_history() is created or replaced with it. Applications only insert into history through the trigger; the trigger itself closes the previous version. A table name whose `_history` would pass PostgreSQL's 63 bytes is refused.

Every version carries the life of its id (life, from 1) and the transaction that wrote it (transaction_id). A row that arrives under an id, by an insert or by an update that changes its id onto it, starts that id's next life: a re-create after a delete or a TRUNCATE, a ClientID create of an old id (D18) and an id change onto an id that lived before never continue the old item's history. Only the insert half of a partition move, whose id's last version is a delete of the same transaction, continues the life, so a moved row keeps its whole history; a delete and a re-create of one id inside one transaction cannot be told from a move and stay one life too. An id change starts the new id's history at the change; the old versions stay under the old id. Every other version continues the life of the one before. HistoryPage reads the current life only (D56).

func ListPage ¶

func ListPage[Item any](ctx context.Context, database DBTX, specification *query.ResourceSQL, listQuery *query.ListQuery, hydrate Hydrate[Item]) (*livewire.Page[Item], error)

ListPage reads one page of a filterable resource (section 8, D53, D55) as the resource's List loader answers it: the page query, the capped count and the hydration of the page's rows, all on one snapshot; the items are in page order (never nil) and Next is set when the page is full. Inside an engine unit of work (TransactionFrom(ctx) is not nil) they run in the engine's transaction, whatever database is, so the read sees the engine's snapshot and livewire.tenant; outside one, on a pool or a connection, ListPage opens a REPEATABLE READ READ ONLY transaction of its own. A cursor value the column refuses (SQLSTATE 22P02, a forged enum value) is a *query.Error with query.CodeInvalidCursor; an id the hydration did not return is an error.

package main

import (
	"context"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5/pgxpool"

	"github.com/govengers/livewire"
	"github.com/govengers/livewire/postgresadapter"
	"github.com/govengers/livewire/query"
)

func main() {
	type Part struct {
		ID   uuid.UUID
		Name string
	}
	partSQL := &query.ResourceSQL{From: "parts", IDColumn: "parts.id"}
	hydrate := func(ctx context.Context, database postgresadapter.DBTX, rows []postgresadapter.PageRow) (map[uuid.UUID]Part, error) {
		parts := map[uuid.UUID]Part{}
		result, err := database.Query(ctx, "SELECT id, name FROM parts WHERE id = ANY($1)", postgresadapter.PageIDs(rows))
		if err != nil {
			return nil, err
		}
		defer result.Close()
		for result.Next() {
			var part Part
			if err := result.Scan(&part.ID, &part.Name); err != nil {
				return nil, err
			}
			parts[part.ID] = part
		}
		return parts, result.Err()
	}
	var pool *pgxpool.Pool // the repository's pool
	listParts := func(ctx context.Context, listQuery *query.ListQuery) (*livewire.Page[Part], error) {
		return postgresadapter.ListPage(ctx, pool, partSQL, listQuery, hydrate) // ResourceOptions.List
	}
	_ = listParts
}

Output:

func PageIDs ¶

func PageIDs(rows []PageRow) []uuid.UUID

PageIDs is the ids of rows in page order, the argument of a hydration query.

func PartitionedTriggerSQL ¶

func PartitionedTriggerSQL(table, column string) string

PartitionedTriggerSQL is TriggerSQL for one table whose changes carry the value of column as their partition (D25), so a change re-runs only the subscriptions of that partition and those without one.

func TriggerSQL ¶

func TriggerSQL(tables ...string) string

TriggerSQL makes every table notify on each row change and on TRUNCATE (D22); the changes carry no partition, so they reach the subscriptions of every partition. A table name is one identifier in the search path, quoted as it is. On a natively partitioned table the row trigger reaches every partition and the changes name the table itself; the TRUNCATE trigger is installed on the table and on the partitions it has now, so a partition attached later notifies its own TRUNCATE only once TriggerSQL runs again, and a DETACH PARTITION notifies nothing.

func WriteCSV ¶

func WriteCSV(ctx context.Context, database DBTX, export *Export, writer io.Writer) (int, error)

WriteCSV streams every row under export.Query's WHERE, in its order, as CSV to writer: a header of the column names, then one record per row, read in chunks by keyset so no offset grows; Limit and Offset of the query are ignored. A column whose Read check refuses the identity in ctx is left out, header included (D39). It runs on one snapshot like ListPage, in the engine's transaction inside an engine unit of work, and returns how many rows it wrote. Values are written as text: times in RFC 3339 UTC, uuids hyphenated, NULL as an empty field, arrays and json as JSON. A text value a spreadsheet would run as a formula, one that starts with =, +, -, @, a tab or a carriage return, is written with a single quote in front (D69), so opening the export runs nothing, and so is one that starts with a quote, so a reader drops exactly one leading quote of every text cell that starts with one; numbers, times and every other value are written by their type and keep their form.