NAME

EV::Pg - asynchronous PostgreSQL client using libpq and EV

SYNOPSIS

use v5.10;
use EV;
use EV::Pg;

my $pg = EV::Pg->new(
    conninfo   => 'dbname=mydb',
    on_error   => sub { warn "PG error: $_[0]\n"; EV::break },
);
$pg->on_connect(sub {
    $pg->query_params(
        'select $1::int + $2::int', [10, 20],
        sub {
            my ($rows, $err) = @_;
            if ($err) { warn $err; EV::break; return; }
            say $rows->[0][0];  # 30
            EV::break;
        },
    );
});
EV::run;

DESCRIPTION

EV::Pg is a non-blocking PostgreSQL client built on top of libpq and the EV event loop. It drives the libpq async API (PQsendQuery, PQconsumeInput, PQgetResult) through ev_io watchers on the libpq socket, so the event loop never blocks on database I/O.

Features: parameterized queries, prepared statements, pipeline mode, single-row mode, chunked rows (libpq >= 17), COPY IN/OUT, LISTEN/NOTIFY, async cancel (libpq >= 17), structured error fields, protocol tracing, and notice handling.

CALLBACKS

Query callbacks always receive a single positional argument on success and (undef, $error_message) on error, so

my ($result, $err) = @_;

works for every shape: $result is the success payload, $err is defined only on error. The shape of $result depends on the query:

SELECT (or single-row / chunked mode)

\@rows -- arrayref of rows; each row is an arrayref of column values with SQL NULL mapping to Perl undef.

INSERT / UPDATE / DELETE

$cmd_tuples -- the string from PQcmdTuples (e.g. "1", "0").

PREPARE / close_prepared / close_portal

"" -- always an empty string (these commands return no row count).

describe_prepared / describe_portal

\%meta -- hashref with nfields, nparams, and (when non-zero) fields (arrayref of {name, type} hashes) and paramtypes (arrayref of OIDs).

COPY

"COPY_IN", "COPY_OUT", or "COPY_BOTH" -- a string tag identifying the COPY direction.

pipeline_sync

1.

Exceptions thrown inside callbacks are caught and reported via warn so that one bad callback does not derail the rest of the queue.

All string values are byte strings as returned by libpq, never decoded characters; decode them yourself (see client_encoding).

A callback may nest EV::run, but while a query callback or on_notice runs, the object's I/O waits for it to return: never wait there for pg results, the wait would never end. Nested waits from on_connect, on_notify and on_drain work.

CONSTRUCTOR

new

my $pg = EV::Pg->new(%args);

Returns a new EV::Pg object. If conninfo or conninfo_params is supplied, an asynchronous connect starts immediately; otherwise call connect later.

Recognized arguments:

conninfo

libpq connection string passed to connect.

conninfo_params

Hashref of connection parameters (e.g. { host => 'localhost', dbname => 'mydb', port => 5432 }), passed to connect_params. Takes precedence over conninfo when both are given.

expand_dbname

When true together with conninfo_params, the dbname value is itself parsed as a connection string -- so dbname => 'postgresql://host/db?sslmode=require' works. Ignored unless conninfo_params is also given.

on_connect

Fires with no arguments on every successful handshake, including each reset.

on_error

Fires as ($error_message) on connection-level errors. The default handler dies, which, like any handler exception, is caught and reported via warn. on_error => undef still installs the default; clear it with $pg->on_error(undef).

on_notify

Fires as ($channel, $payload, $backend_pid) for LISTEN/NOTIFY messages.

on_notice

Fires as ($message) for server NOTICE/WARNING messages.

on_drain

Fires with no arguments when the libpq send buffer has been fully flushed during a COPY -- use it to resume sending after put_copy_data returned 0.

keep_alive

When true, the connection keeps EV::run alive even with an empty callback queue. See "keep_alive".

loop

An EV loop object. Defaults to EV::default_loop.

Unknown arguments produce a carp warning and are otherwise ignored.

OBJECT LIFECYCLE

Destroying the object (undef $pg, including from inside a callback) fires every queued callback with (undef, "object destroyed") and closes the connection.

Handlers that close over $pg form a reference cycle: only undef $pg (or clearing the handlers) frees the object; scope exit and finish do not.

After fork, the child inherits a copy it must not use for I/O. Destroying it (undef $pg or plain exit), finish and skip_pending are safe: the parent's connection is untouched and inherited callbacks are dropped silently. Everything that talks to the server croaks on the inherited object. Destroy inherited objects before running an event loop in the child, and create a new object there to talk to PostgreSQL.

CONNECTION METHODS

connect

$pg->connect($conninfo);

Starts an asynchronous connection from a libpq connection string. on_connect fires on success, on_error on failure. Croaks if already connected, and synchronously for failures detectable before the first packet (bad keyword, unresolvable host, missing Unix socket, on some systems a refused local port); the rest arrive via on_error. Do not call connect directly from on_error: the failed connection is still installed while the handler runs, so it croaks -- finish first, or defer with a timer like reset.

connect_params

$pg->connect_params(\%params);
$pg->connect_params(\%params, $expand_dbname);

Like connect but takes a hashref of keyword/value parameters. When $expand_dbname is true, the dbname entry may itself be a connection string or URI. Undef values mean the libpq default. Croaks if already connected, or synchronously if the parameters are unusable.

reset

$pg->reset;

Drops the current connection and reconnects with the same parameters. Pending callbacks fire with (undef, "connection reset") first. Croaks if no connection was ever started. Alias: reconnect.

Do not call reset directly from on_error against a possibly-down server: each attempt fails immediately and reconnects spin hot. Defer with a timer, and catch the synchronous failures connect lists:

on_error => sub {
    my $t; $t = EV::timer(5, 0, sub {
        undef $t;
        eval { $pg->reset; 1 } or $pg->on_error->($@);
    });
},

finish

$pg->finish;

Closes the connection. Pending callbacks fire with (undef, "connection finished"). Alias: disconnect.

is_connected

my $bool = $pg->is_connected;

True if the handshake has completed and the connection is ready for queries. False during connect, after finish, and after a connection-level failure. A failed query (PGRES_FATAL_ERROR) leaves the connection usable, so it does not affect this flag.

status

my $st = $pg->status;

libpq connection status: CONNECTION_OK or CONNECTION_BAD; intermediate libpq states may show during the handshake. Returns CONNECTION_BAD when not connected.

QUERY METHODS

The methods below that issue commands croak when called while not connected; those taking a callback also croak when it is not a coderef. Outside pipeline mode only one query may be in flight; sending another before the first completes croaks.

query

$pg->query($sql, sub { my ($result, $err) = @_; });

Sends a simple query. Multi-statement strings (e.g. "SELECT 1; SELECT 2") are accepted, but only the final result reaches the callback -- intermediate results are silently discarded, and because PostgreSQL stops at the first error, errors always arrive as that final result. Not allowed in pipeline mode -- use query_params there. Alias: q. An empty, whitespace-only or semicolon-only string yields (undef, "empty query").

query_params

$pg->query_params($sql, \@params, sub { my ($result, $err) = @_; });

Sends a parameterized query. Parameters are referenced in SQL as $1, $2, etc.; undef elements become SQL NULL.

Values are sent in PostgreSQL's text format. Embedded NUL bytes cause the call to croak (text-format params cannot legally contain NULs) -- pass binary data through escape_bytea if you need a bytea column. The params argument must be an arrayref. Alias: qp.

prepare

$pg->prepare($name, $sql, sub { my ($result, $err) = @_; });

Creates a prepared statement at the protocol level (no SQL PREPARE parsing). The callback receives "" on success. Alias: prep.

query_prepared

$pg->query_prepared($name, \@params, sub { my ($result, $err) = @_; });

Executes a prepared statement created by prepare. Same param rules as query_params. Alias: qx.

describe_prepared

$pg->describe_prepared($name, sub { my ($meta, $err) = @_; });

Describes a prepared statement. The callback receives a hashref with keys nfields and nparams. When nfields is non-zero, a fields key is also present (arrayref of {name, type} hashes). When nparams is non-zero, a paramtypes key is also present (arrayref of OIDs).

describe_portal

$pg->describe_portal($name, sub { my ($meta, $err) = @_; });

Describes a portal. The callback receives the same hashref structure as describe_prepared.

set_single_row_mode

my $ok = $pg->set_single_row_mode;

Switches the most recently sent query into single-row mode. Must be called immediately after a send method (query, query_params, ...) and before the event loop delivers any results -- a 0 return means no query was in the right async state and should be treated as a programmer error rather than a runtime condition.

The query callback then fires once per row with a single-row \@rows (e.g. [[$col0, $col1, ...]]), and once more at the end with an empty \@rows as the completion sentinel.

set_chunked_rows_mode

my $ok = $pg->set_chunked_rows_mode($chunk_size);

Like set_single_row_mode but delivers up to $chunk_size rows per callback (requires libpq >= 17), reducing per-callback overhead for large result sets. Same call-timing constraint and same trailing empty-rows completion sentinel.

close_prepared

$pg->close_prepared($name, sub { my ($result, $err) = @_; });

Closes (deallocates) a prepared statement at protocol level (requires libpq >= 17). The callback receives an empty string ("") on success. Works in pipeline mode, unlike DEALLOCATE SQL.

close_portal

$pg->close_portal($name, sub { my ($result, $err) = @_; });

Closes a portal at protocol level (requires libpq >= 17). The callback receives an empty string ("") on success.

cancel

my $err = $pg->cancel;

Sends a cancel request using the legacy PQcancel API. Blocks the event loop for one network round trip; prefer cancel_async on libpq >= 17. Returns undef on success or an error string on failure.

cancel_async

$pg->cancel_async(sub { my ($r, $err) = @_; });

Sends a non-blocking cancel request using the PQcancelConn API (requires libpq >= 17). The callback receives (1) on success or (undef, $errmsg) on failure. Croaks if a cancel is already in progress. Success means the cancel was sent, not that the query died: a late cancel loses and the query callback still delivers its result. There is no timeout; guard with your own timer if the server may be unreachable.

pending_count

my $n = $pg->pending_count;

Number of callbacks currently in the queue (queries sent but not yet delivered).

keep_alive

$pg->keep_alive(1);
my $bool = $pg->keep_alive;

When true, the read watcher keeps EV::run alive even when the callback queue is empty. Required when waiting for server-side NOTIFY events via on_notify -- without this flag the loop would exit as soon as the LISTEN query completes. Getter/setter.

skip_pending

$pg->skip_pending;

Drops every queued callback, invoking each with (undef, "skipped"). Results still in flight are discarded as they arrive; until then a send outside pipeline mode croaks with another command is already in progress, so retry it later. A skipped COPY OUT or COPY BOTH is ended and drained automatically; a skipped COPY IN is aborted when skipped from its own callback, otherwise end it with put_copy_end (or reset).

PIPELINE METHODS

Pipeline mode lets you send multiple queries without waiting for individual results, then receive the results in order after a sync point. Inside a pipeline you must use query_params or query_prepared -- query is rejected.

enter_pipeline

$pg->enter_pipeline;

Switches the connection into pipeline mode. Croaks if there are unfinished results outstanding. Do not issue COPY ... FROM STDIN in pipeline mode: an unfed COPY stalls, and further queries make the server abort the connection (COPY ... TO STDOUT is fine). If a query in a batch fails, the server skips the rest of the batch and their callbacks fire with (undef, "pipeline aborted").

exit_pipeline

$pg->exit_pipeline;

Returns to normal mode. Croaks if the pipeline is not idle. Succeeds quietly when not in pipeline mode.

pipeline_sync

$pg->pipeline_sync(sub { my ($r, $err) = @_; });

Sends a pipeline sync point. The callback fires with (1) after all preceding queries in the batch have completed, or (undef, $errmsg) if the connection drops first. Croaks outside pipeline mode. Alias: sync.

send_pipeline_sync

$pg->send_pipeline_sync(sub { my ($r, $err) = @_; });

Like pipeline_sync but does not flush the send buffer (requires libpq >= 17). Useful for batching multiple sync points before a single manual flush via send_flush_request. Croaks outside pipeline mode.

send_flush_request

$pg->send_flush_request;

Asks the server to deliver results for queries sent so far -- the manual companion to send_pipeline_sync. Alias: flush.

pipeline_status

my $st = $pg->pipeline_status;

One of PQ_PIPELINE_OFF, PQ_PIPELINE_ON, or PQ_PIPELINE_ABORTED.

COPY METHODS

During a COPY the query callback first fires with a string tag ("COPY_IN" / "COPY_OUT" / "COPY_BOTH") when streaming starts -- for COPY OUT and BOTH again whenever more data arrives -- and once more with the final command result (or error) when the stream ends. See eg/copy_in.pl and eg/copy_out.pl.

put_copy_data

my $rc = $pg->put_copy_data($data);

Sends a chunk during COPY IN. Returns 1 on success (data buffered or flushed), or -1 on error; -1 also outside COPY IN state. libpq grows its send buffer without limit, so pace large loads yourself (e.g. via on_drain). A 0 means the buffer could not grow: wait for on_drain, then retry.

put_copy_end

my $rc = $pg->put_copy_end;
my $rc = $pg->put_copy_end($errmsg);

Ends a COPY IN. With $errmsg aborts the COPY server-side. Same return convention as put_copy_data (including -1 outside COPY IN state).

get_copy_data

my $row = $pg->get_copy_data;

Retrieves the next row during COPY OUT. Returns the row bytes, the integer -1 when the stream is complete, or undef if nothing is currently buffered. After undef, return from the callback: it fires again with "COPY_OUT" when more data arrives. Croaks when called outside COPY OUT state.

HANDLER METHODS

Each handler is a getter/setter: pass a coderef to install it (returning the new value), pass undef to clear it, or call without arguments to read the current handler. Passing a defined non-coderef croaks.

on_connect

Fires with no arguments after every successful handshake, including each reset.

on_error

Fires as ($error_message) for connection-level errors (handshake failure, lost socket, libpq protocol errors). Per-query errors come through the query callback, not here.

When the socket is lost, on_error fires with the libpq error message and every callback still outstanding when the loss is detected then fires with (undef, "connection lost") (including pipeline_sync callbacks). A final server error may reach a query callback before on_error fires.

on_notify

Fires as ($channel, $payload, $backend_pid) for each LISTEN/NOTIFY message. Without a handler, notifications are silently discarded.

on_notice

Fires as ($message) for server NOTICE/WARNING messages. Without a handler, notices are silently discarded (this overrides libpq's stderr default). The handler runs inside libpq, so the COPY methods, set_client_encoding and encrypt_password without an algorithm croak while it runs.

on_drain

Fires with no arguments when the libpq send buffer has been fully flushed during a COPY -- use it to resume put_copy_data after a 0 return.

CONNECTION INFO

String accessors (db, user, host, hostaddr, port, error_message, parameter_status, ssl_attribute) return undef when not connected. Integer accessors return a default value (0, -1, or the enum's unknown member). Methods that require an active connection (client_encoding, set_client_encoding, set_error_verbosity, set_error_context_visibility, conninfo) croak otherwise.

error_message

my $msg = $pg->error_message;

Last error message from libpq. Alias: errstr.

transaction_status

my $st = $pg->transaction_status;

One of PQTRANS_IDLE, PQTRANS_ACTIVE, PQTRANS_INTRANS, PQTRANS_INERROR, PQTRANS_UNKNOWN. Alias: txn_status.

parameter_status

my $val = $pg->parameter_status($name);

Server parameter value (e.g. "server_version", "client_encoding", "server_encoding").

backend_pid

my $pid = $pg->backend_pid;

Backend process ID. Alias: pid.

server_version

my $ver = $pg->server_version;

Server version as a packed integer: MMmmpp (major * 10000 + minor * 100 + patch) for releases before 10, MM0000 + patch from 10 onwards. PostgreSQL 18.0 returns 180000; 17.5 returns 170005.

protocol_version

my $ver = $pg->protocol_version;

Frontend/backend protocol version (typically 3).

db

my $dbname = $pg->db;

Database name.

user

my $user = $pg->user;

Connected user name.

host

my $host = $pg->host;

Server host as supplied to connect (may be a hostname or socket dir).

hostaddr

my $addr = $pg->hostaddr;

Server IP address (empty string over a Unix socket).

port

my $port = $pg->port;

Server port.

socket

my $fd = $pg->socket;

Underlying socket file descriptor (for advanced uses such as installing your own watcher). Returns -1 when not connected.

ssl_in_use

my $bool = $pg->ssl_in_use;

True if the connection is encrypted with SSL.

ssl_attribute

my $val = $pg->ssl_attribute($name);

SSL attribute (e.g. "protocol", "cipher", "key_bits").

ssl_attribute_names

my $names = $pg->ssl_attribute_names;

Arrayref of available SSL attribute names, or undef if the connection does not use SSL.

client_encoding

my $enc = $pg->client_encoding;

Current client encoding name (e.g. "UTF8").

set_client_encoding

$pg->set_client_encoding($encoding);

Sets the client encoding (e.g. "UTF8", "SQL_ASCII"). This is a synchronous (blocking) call that stalls the event loop for one server round trip, so it is best invoked right after on_connect fires and before any queries are dispatched. Croaks if there are pending queries or on failure. Notifications read during the call are delivered to on_notify before it returns.

set_error_verbosity

my $old = $pg->set_error_verbosity($level);

Sets error verbosity. $level is one of PQERRORS_TERSE, PQERRORS_DEFAULT, PQERRORS_VERBOSE, or PQERRORS_SQLSTATE. Returns the previous setting.

set_error_context_visibility

my $old = $pg->set_error_context_visibility($level);

Sets error context visibility. $level is one of PQSHOW_CONTEXT_NEVER, PQSHOW_CONTEXT_ERRORS (default), or PQSHOW_CONTEXT_ALWAYS. Returns the previous setting.

error_fields

my $fields = $pg->error_fields;

Returns a hashref of structured error fields from the most recent PGRES_FATAL_ERROR result, or undef if no fatal error has been seen. Persists until the next fatal error; successful queries do not clear it, but reset/finish do. Each key is present only when the corresponding field is non-NULL in the server response:

sqlstate            severity            primary
detail              hint                position
context             schema              table
column              datatype            constraint
internal_position   internal_query
source_file         source_line         source_function

result_meta

my $meta = $pg->result_meta;

Returns a hashref of metadata for the most recent query result, or undef if no result has been delivered. Refreshed by every successful result (including commands with no columns, and the COMMAND_OK that terminates a COPY) but not by errors, COPY tags while streaming, or pipeline sync results. Reliable only inside the callback of the result it describes: after an error, COPY tag or sync it may return undef or the previous result's metadata. Cleared by reset/finish. The most recent result is retained until the next one arrives; for very large results prefer single-row or chunked mode.

Keys:

nfields       number of columns
cmd_status    command status string (e.g. "SELECT 3", "INSERT 0 1")
inserted_oid  OID of inserted row -- only present for single-row
                INSERTs that generated an OID (legacy WITH OIDS
                tables); absent for normal INSERTs and other commands
fields        arrayref of column metadata hashrefs:
                name, type (OID), ftable (OID), ftablecol,
                fformat (0=text, 1=binary), fsize, fmod

conninfo

my $info = $pg->conninfo;

Returns a hashref of the connection parameters actually used by the live connection (keyword => value pairs). Warning: includes the password in cleartext when the connection string has one; do not log it wholesale.

connection_used_password

my $bool = $pg->connection_used_password;

Returns 1 if the connection authenticated with a password.

connection_used_gssapi

my $bool = $pg->connection_used_gssapi;

Returns 1 if the connection used GSSAPI authentication. Requires libpq >= 16 (method not defined below that).

connection_needs_password

my $bool = $pg->connection_needs_password;

Returns 1 if the server requested a password during authentication.

trace

$pg->trace($filename);

Enables libpq protocol tracing, writing the wire-level frontend/backend exchange to $filename (truncating it). Croaks if the file cannot be opened, or when not connected. Tracing stops if the connection is reset, finished, or lost; call trace again after reconnecting.

untrace

$pg->untrace;

Stops tracing and closes the trace file. Safe to call when tracing is not active.

set_trace_flags

$pg->set_trace_flags($flags);

Sets the trace output style. $flags is a bitmask of PQTRACE_SUPPRESS_TIMESTAMPS and/or PQTRACE_REGRESS_MODE (handy when diffing traces). Croaks when not connected.

UTILITY METHODS

escape_literal, escape_identifier, escape_bytea and encrypt_password require a live connection and croak otherwise.

escape_literal

my $quoted = $pg->escape_literal($string);

Quotes and escapes a string for safe interpolation into SQL (wraps the value in single quotes and doubles internal quotes). Alias: quote.

escape_identifier

my $quoted = $pg->escape_identifier($string);

Quotes and escapes an identifier for safe interpolation into SQL (wraps in double quotes). Alias: quote_id.

escape_bytea

my $escaped = $pg->escape_bytea($binary);

Escapes binary bytes into the textual bytea form expected by the server (the \x... hex notation). Pair with unescape_bytea to go the other way.

encrypt_password

my $hashed = $pg->encrypt_password($password, $user);
my $hashed = $pg->encrypt_password($password, $user, $algorithm);

Hashes a password client-side (so the cleartext never reaches the server) ready to be passed to ALTER ROLE ... PASSWORD. $algorithm is optional; when omitted the server's password_encryption setting decides (typically "scram-sha-256"). That lookup blocks like set_client_encoding and croaks with pending or undrained queries; pass an explicit algorithm to avoid the server round trip.

unescape_bytea

my $binary = EV::Pg->unescape_bytea($escaped);

Class method. Decodes the textual bytea form back to raw bytes.

lib_version

my $ver = EV::Pg->lib_version;

Class method. Returns the libpq version as an integer (same encoding as server_version; e.g. 170000 for libpq 17.0).

conninfo_parse

my $params = EV::Pg->conninfo_parse($conninfo);

Class method. Parses a connection string and returns a hashref of the recognized keyword/value pairs, croaking if the string is malformed. Handy for validating connection strings before connecting.

ALIASES

Short aliases for common methods:

q           query
qp          query_params
qx          query_prepared
prep        prepare
reconnect   reset
disconnect  finish
flush       send_flush_request
sync        pipeline_sync
quote       escape_literal
quote_id    escape_identifier
errstr      error_message
txn_status  transaction_status
pid         backend_pid

EXPORT TAGS

:status       PGRES_* result status constants
:conn         CONNECTION_OK, CONNECTION_BAD
:transaction  PQTRANS_* transaction status constants
:pipeline     PQ_PIPELINE_* pipeline status constants
:verbosity    PQERRORS_* verbosity constants
:context      PQSHOW_CONTEXT_* context visibility constants
:trace        PQTRACE_* trace flag constants
:all          all of the above

BENCHMARK

500k queries over Unix socket, PostgreSQL 18, libpq 18:

Workload   EV::Pg sequential  EV::Pg pipeline  DBD::Pg sync  DBD::Pg async+EV
SELECT          83,998 q/s      144,939 q/s     73,195 q/s      65,966 q/s
INSERT          67,053 q/s       85,701 q/s     60,127 q/s      58,329 q/s
UPSERT          37,360 q/s       43,019 q/s     40,278 q/s      40,173 q/s

Sequential mode uses prepared statements (parse once, bind+execute per call). Pipeline mode batches queries with pipeline_sync every 1000 queries. Reproduce with bench/bench.pl (BENCH_N queries per workload, default 10,000; BENCH_BATCH pipeline batch size, default 1000).

REQUIREMENTS

libpq >= 14 (PostgreSQL client library) and EV. A handful of features -- chunked rows mode, close_prepared/close_portal, send_pipeline_sync, and cancel_async -- require libpq >= 17, connection_used_gssapi needs >= 16. Built against older headers these methods are not defined; run against an older library they croak.

THREADS

EV::Pg is not thread-safe and must not be used across Perl ithreads, even with separate connections per thread: internal state is unsynchronized and EV watchers are single-threaded. Use separate processes for parallelism. Spawning a thread while an object is visible is safe -- the child gets an unusable unblessed copy and the parent's connection is unaffected.

SEE ALSO

EV, DBD::Pg, Mojo::Pg, AnyEvent::Pg

LICENSE

This is free software; you can redistribute it and/or modify it under the same terms as the Perl 5 programming language system itself.

cpanm EV::Pg

perl -MCPAN -e shell
install EV::Pg