pg

A PostgreSQL client, written in slang over net. A query waiting on the server parks its task like any socket read, so unlike sql it does not block a worker thread: a server can hold many database connections on a handful of threads. Errors keep the server's own text and SQLSTATE, through the same result[_, str] story as every other package.

import "pg";
import "time";
import "log";

let dl = until_of(time.mono() + 5000000000);

let pr = pg.new_pool("postgres://app:secret@db.internal/shop", 10);
guard let pool = pr else let e = err_of(pr) {
    log.error("db url: " + e);
    exit(1);
}

let r = pg.pool_query(pool,
    "SELECT id, name, price FROM products WHERE price < $1 ORDER BY id",
    [pg.arg_float(20.0)], dl);
guard let rows = r else let e = err_of(r) {
    log.error("query: " + e);   // ERROR: relation "products" does not exist (SQLSTATE 42P01)
    exit(1);
}
let i = 0;
while i < rows.count {
    println(to_str(pg.get_int(rows, i, 0)) + " " + pg.get_text(rows, i, 1));
    i = i + 1;
}

Every call that talks to the server takes a deadline (until), like httpc; until_of(0) waits indefinitely. For connect it covers the whole connection: DNS lookup, TCP connect, TLS handshake and login.

Parameters travel separately from the SQL text, so a value can never be parsed as SQL, whatever it contains. Build them with pg.arg_text(s), arg_int(n), arg_float(x) (sent exactly, not rounded), arg_bool(b), arg_bytes(b) (binary, for bytea), arg_time(t) (a timez.Time, for timestamptz; Postgres keeps microseconds) and arg_null(). A query with none takes [].

Results from query are buffered whole in a Rows (for one too big for that, see Streaming): rows.count, rows.columns (names), rows.affected (from the command tag, so an INSERT or UPDATE through query reports its row count) and rows.tag. Read cells by row and column index; pg.col(rows, "name") finds an index.

The getters panic on NULL, on a column of a type they do not read, on an unknown column name and on an index out of range, with a message naming the column (column 'email' is NULL in row 3; check pg.is_null before pg.get_text). Those are disagreements between the query and the code reading it — bugs to see, not conditions to handle — and inventing a 0 or "" would hide them. A nullable column is checked with is_null first. Like sqlx's get, not Go's Scan.

pg.rows_as[T](rows) -> result[[T], str] binds columns by exact field name and decodes directly into value or gc structs, without rendering rows as JSON. Annotate the result, for example:

struct User { id: int, email: str, nickname: opt[str] }
let converted: result[[User], str] = pg.rows_as(rows);

Extra columns are ignored. Missing optional fields and SQL NULL become none; missing required fields and incompatible cells return an error naming the row and field. Binary integers and booleans retain their types, numeric values retain integer precision, and JSON/JSONB columns decode into nested types. bytea reads into bytes (or a base64 str); timestamps read into timez.Time or RFC 3339 strings, with a zone-free timestamp read as UTC. Infinity and BC timestamps remain strings and cannot fit a Time. String-keyed maps and optional row wrappers are also supported. Duplicate names use the first column for structs and the last for maps, as JSON decoding does. Scalar row targets are rejected; declare a row struct.

The hermetic bench/pg_rows_as/main.sl compares this with the previous Zokor JSON adapter on the same 1,000-row fixture. Intel macOS, ABBA x4: 3.792 ms versus 0.263 ms median per conversion; 23,040 versus 3,671 allocations and 1,864,778 versus 255,780 allocated bytes per conversion. These measure mapping alone, excluding database and request latency.

Pool#

pg.new_pool(url, max_open) parses the url and connects nothing until first use. pool_query and pool_exec take a connection, run, and give it back, so they cannot leak one. A transaction needs several statements on one connection, so it uses acquire and release:

let cr = pg.acquire(pool, dl);
guard let c = cr else let e = err_of(cr) { return; }
pg.exec(c, "BEGIN", dl);
let moved = pg.query(c, "UPDATE accounts SET balance = balance - $1 WHERE id = $2",
                     [pg.arg_int(100), pg.arg_int(from)], dl);
guard let m = moved else {
    pg.release(pool, c);    // still in the transaction: closed, never reused
    return;
}
pg.exec(c, "COMMIT", dl);
pg.release(pool, c);

At most max_open connections exist at once; a task that needs one while all are in use waits, up to its deadline. Waiters are served in arrival order: a release hands its connection straight to the oldest one, and pool_close fails every acquire still waiting. Idle connections are probed before reuse and closed after pool.idle_timeout (5 minutes). A connection released while broken, closed or still inside a transaction is closed rather than handed on — the next caller would otherwise run its statements inside someone else's uncommitted work. Releasing the same connection twice panics. pool.dials and pool.reuses count what the pool did.

URLs and TLS#

postgres://user:password@host:port/database?param=value, with % escapes in any part. The port defaults to 5432 and the database to the user name. Recognised parameters are sslmode, sslrootcert (a CA bundle to verify against, for providers that sign with their own CA), application_name and statement_cache_capacity (below), plus host and port, which override the ones in the authority, as in libpq. Anything else is an error, not ignored: a misspelt sslmdoe=require that was quietly dropped would connect in cleartext and nothing would ever say so.

This differs from libpq on purpose: libpq's require encrypts without checking who answered. Without sslmode, the default is require — except for a loopback host (localhost, 127.x, ::1), where it is disable, because traffic that never leaves the machine gains nothing from TLS and local servers rarely have it set up. A server that does not offer TLS gets a message saying to add sslmode=disable.

Unix-domain sockets are named by their directory, as libpq does: postgres://app@%2Fvar%2Frun%2Fpostgresql/shop (the host percent-encoded) or postgres://app@/shop?host=/var/run/postgresql. The socket file is <dir>/.s.PGSQL.<port>. Postgres does not use TLS over a socket, so sslmode defaults to disable there and require is an error.

Authentication: SCRAM-SHA-256 (the default since Postgres 14), md5 and cleartext password. SCRAM checks that the server knows the password too: a server that skips its proof, or sends a wrong one, is refused.

Timeouts and errors#

A server-side error — a syntax error, a constraint violation — leaves the connection usable. Anything that leaves the stream at an unknown point breaks it: an I/O error, a protocol violation, a deadline. After a "timeout" the server may still be running the query, so the driver asks it to stop with a cancel request on a separate connection, from its own task, and returns immediately. A broken connection only returns connection is broken: <why>.

let r = pg.query(c, "INSERT INTO users (email) VALUES ($1)", [pg.arg_text(email)], dl);
guard let ok_ = r else let e = err_of(r) {
    if pg.sqlstate(e) == "23505" {     // unique_violation
        return respond(409, "email already registered");
    }
    return respond(500, "database error");
}

A Conn is safe to share between tasks (calls take turns), but a pool is the better tool for that.

Streaming#

pg.stream runs a query like query but hands the rows over one at a time, so memory holds one row however large the result, and the server is held back by TCP flow control rather than the client reading ahead:

let sr = pg.stream(c, "SELECT id, body FROM events ORDER BY id", [], dl);
guard let rows = sr else let e = err_of(sr) { return; }
while true {
    let nr = pg.next_row(c, rows, dl);
    guard let more = nr else let e = err_of(nr) { log.error(e); break; }
    if !more { break; }
    archive(pg.get_int(rows, 0, 0), pg.get_text(rows, 0, 1));   // always row 0
}

Each next_row replaces the row in rows, read at index 0; after the last, rows.affected and rows.tag are set. A server error part way through comes from next_row, after the rows before it. While a stream is open the connection serves nothing else — other calls fail with connection is busy — so read it to the end or call pg.stream_close(c, rows, dl). That asks the server to cancel the query, waits until the request has been delivered (so it cannot hit a later query on the same connection), then drops whatever was already sent. A connection released to a pool mid-stream is closed.

COPY#

COPY … FROM STDIN bulk-loads data in COPY's text or CSV format, far faster than one INSERT per row; COPY … TO STDOUT dumps it.

let n = pg.copy_from(c, "COPY users (id, email) FROM STDIN (FORMAT csv)",
                     to_bytes("1,a@example.com\n2,b@example.com\n"), dl);
let dump = pg.copy_to(c, "COPY users TO STDOUT (FORMAT csv)", dl);

The start/send/end and start/next forms stream, for data that should not be held in memory at once. A bad row fails the whole COPY — the error comes from copy_in_end (or copy_from) with its SQLSTATE, and nothing was loaded. A COPY of the wrong direction is refused with not a COPY … FROM STDIN statement, and a COPY through query or exec fails with a message pointing here, the connection intact either way. Only the text and CSV formats are meant: binary COPY data passes through untouched but this package does not build or parse it.

LISTEN / NOTIFY#

pg.listen(c, "jobs", dl);
while true {
    let wr = pg.wait_notification(c, until_of(time.mono() + 30000000000));
    guard let got = wr else let e = err_of(wr) { log.error(e); break; }
    guard let note = got else { continue; }        // 30s with nothing: none
    run_job(note.payload);
}

Channel names are quoted as identifiers, so any name is safe to pass. Notifications that arrive during other calls — even in the middle of a query's result — are queued on the connection (up to 10,000; c.notes_dropped counts any beyond that) and handed out oldest first. Reaching the deadline in wait_notification returns none and leaves the connection usable, unlike every other call: waiting is the point, and a message cut off by the deadline is kept and completed by the next wait. A server that ends the session while one waits (an administrator's pg_terminate_backend, a restart) is an error with its SQLSTATE. Listen on a connection of its own, not a pooled one: a pooled connection with a notification waiting fails the pool's idle probe and is closed.

Limits#

A driver trusts the server with its memory, so what it will buffer is capped: one protocol message at 256 MiB, and one result at 256 MiB of cell data, which is an error rather than an allocation. A SCRAM server may ask for at most 1,000,000 PBKDF2 iterations (Postgres uses 4096).

Measured against Postgres 16 in Docker on the development Mac: 1M rows of two columns in about 2.1s and 150MB, a single 20MB value in about 0.3s.

Prepared statements#

query prepares each SQL text once per connection and reuses the statement after that, so Postgres parses and plans it once instead of on every call: on a point read that is about 60% less database CPU. A first use costs no extra round trip (the Parse travels with the query). Each connection keeps its 256 most recently used statements; statement_cache_capacity=N in the url changes that, and 0 turns it off, which a pooler that cannot carry named statements between server connections needs (PgBouncer before 1.21 in transaction mode). On cached runs, int2, int4, int8, bool, and bytea results use PostgreSQL's binary format; the pg.get_* functions preserve their existing values and text representation. Other result types stay in text format. If the server drops a statement or a schema change alters what it returns, the driver prepares it again and retries the query once, outside a transaction; inside one, the error is returned, because the transaction is already aborted. DEALLOCATE ALL and DISCARD ALL are noticed and empty the cache. exec and stream are not prepared.

Not supported: binary floats, building or parsing binary COPY data, Kerberos/GSSAPI, SCRAM channel binding (SCRAM-SHA-256-PLUS), multiple hosts in one url, and SASLprep normalisation of non-ASCII passwords (an ASCII password is unaffected).

API#