Struct Connection 

pub struct Connection { /* private fields */ }

A PostgreSQL connection with statement cache and inline message processing.

Connections are not Send — they must be used on one task at a time. The pool handles concurrent access by lending connections to individual tasks.

Implementations

impl Connection

pub async fn connect(config: &Config) -> Result<Self, DriverError>

Connect to PostgreSQL and complete the startup/auth handshake.

When config.host starts with / (Unix domain socket directory), connects via UnixStream at {host}/.s.PGSQL.{port} instead of TCP. TCP_NODELAY and keepalive are skipped for UDS since they are TCP-only.

pub async fn prepare_only( &mut self, sql: &str, sql_hash: u64, ) -> Result<(), DriverError>

Prepare a statement without executing it (Parse+Describe+Sync only).

Used by connection warmup to pre-cache statements without executing them. If the statement is already cached, this is a no-op.

pub async fn prepare_describe( &mut self, sql: &str, ) -> Result<PrepareResult, DriverError>

Prepare a statement and return full column + parameter metadata.

Sends Parse + Describe(Statement) + Sync, then reads:

• ParseComplete
• ParameterDescription (param type OIDs)
• RowDescription or NoData (column metadata)
• ReadyForQuery

Unlike prepare_only, this always sends Parse (no cache check) and uses the unnamed statement "" so it does not pollute the statement cache. This is designed for compile-time SQL validation in the proc macro, where we need column + param metadata but never execute.

pub async fn simple_query_rows( &mut self, sql: &str, ) -> Result<Vec<SimpleRow>, DriverError>

Execute a simple (text protocol) query and return all result rows.

Each row is a Vec<Option<String>> — NULL values are None, text values are Some(String). This uses the simple query protocol which always returns text-format results.

Designed for compile-time schema introspection queries in the proc macro (e.g. pg_attribute, information_schema). Not intended for high-performance runtime use.

pub async fn query_streaming_start( &mut self, sql: &str, sql_hash: u64, params: &[&(dyn Encode + Sync)], chunk_size: i32, ) -> Result<(Arc<[ColumnDesc]>, bool), DriverError>

Begin a streaming query using the PG extended query protocol with Execute(max_rows=chunk_size).

Returns column metadata and puts the connection into streaming mode. The caller must repeatedly call streaming_next_chunk() until it returns Ok(false) (all rows consumed) before issuing any other query on this connection.

Uses the unnamed portal "" which stays open between Execute calls as long as Sync is NOT sent. We use Flush (not Sync) to force PG to send buffered output without destroying the portal. Sync is only sent after CommandComplete to cleanly end the query cycle.

pub async fn streaming_next_chunk( &mut self, arena: &mut Arena, all_col_offsets: &mut Vec<(usize, i32)>, ) -> Result<bool, DriverError>

Read the next chunk of rows from an in-progress streaming query.

Returns Ok(true) if more rows are available (PortalSuspended), Ok(false) when all rows have been consumed (CommandComplete).

After CommandComplete, this method sends Sync and reads ReadyForQuery, returning the connection to a clean protocol state.

pub async fn streaming_send_execute( &mut self, chunk_size: i32, ) -> Result<(), DriverError>

Send Execute+Flush for the next chunk of a streaming query.

Must be called before streaming_next_chunk() on the 2nd and subsequent chunks (the first chunk’s Execute is sent by query_streaming_start).

Uses Flush (not Sync) to keep the unnamed portal alive.

pub async fn query( &mut self, sql: &str, sql_hash: u64, params: &[&(dyn Encode + Sync)], arena: &mut Arena, ) -> Result<QueryResult, DriverError>

Execute a prepared query and return rows in arena-allocated storage.

If the statement is not yet cached, Parse+Describe+Bind+Execute+Sync are pipelined in a single TCP write. On cache hit, only Bind+Execute+Sync are sent.

pub async fn execute( &mut self, sql: &str, sql_hash: u64, params: &[&(dyn Encode + Sync)], ) -> Result<u64, DriverError>

Execute a query without result rows (INSERT/UPDATE/DELETE).

Skips DataRow parsing entirely — only reads until CommandComplete. Does not allocate an Arena.

pub async fn execute_pipeline( &mut self, sql: &str, sql_hash: u64, param_sets: &[&[&(dyn Encode + Sync)]], ) -> Result<Vec<u64>, DriverError>

Execute the same prepared statement N times with different parameters in a single pipeline round-trip.

Sends all N Bind+Execute messages followed by one Sync. PostgreSQL processes them in order and returns N BindComplete+CommandComplete responses followed by one ReadyForQuery.

This is a real optimization for bulk operations: N inserts in a transaction become 1 round-trip instead of N round-trips.

The statement must already be cached (call execute at least once first, or use prepare_describe). If not cached, it will be prepared inline for the first entry, then the rest use the cached version.

Returns the number of affected rows for each parameter set.

pub async fn for_each<F>( &mut self, sql: &str, sql_hash: u64, params: &[&(dyn Encode + Sync)], f: F, ) -> Result<(), DriverError>

where F: FnMut(PgDataRow<'_>) -> Result<(), DriverError>,

Process each row directly from the wire buffer via a closure.

Zero arena allocation — the closure receives a PgDataRow that reads columns directly from the DataRow message bytes in the read buffer. Column offsets are pre-scanned once per row into a stack-allocated SmallVec.

This is the fastest path for row-by-row processing: no arena, no Vec of offsets, no materialization of the entire result set.

pub async fn for_each_raw<F>( &mut self, sql: &str, sql_hash: u64, params: &[&(dyn Encode + Sync)], f: F, ) -> Result<(), DriverError>

where F: FnMut(&[u8]) -> Result<(), DriverError>,

Process each DataRow as raw bytes — no PgDataRow, no SmallVec, no pre-scanning of column offsets.

The closure receives the raw DataRow message payload (starting with the i16 column count). Generated code decodes columns sequentially inline, advancing a position cursor through the bytes.

This is faster than for_each because it eliminates the SmallVec construction (~20-30ns per row) and the per-column method call overhead.

Optimization: DataRow messages that fit entirely within stream_buf are parsed directly from the buffer (zero-copy — no memcpy into read_buf). Messages that span the buffer boundary fall back to read_message_buffered.

pub async fn simple_query(&mut self, sql: &str) -> Result<(), DriverError>

Simple query protocol — for non-prepared SQL (BEGIN, COMMIT, SET, etc.).

Does not use the extended query protocol. Cannot have parameters.

pub async fn wait_for_notification( &mut self, ) -> Result<(String, String), DriverError>

Block until a NotificationResponse arrives on this connection.

Reads raw messages from the stream and skips everything except NotificationResponse. Returns the (channel, payload) pair. Used by the listener’s background task to receive LISTEN/NOTIFY events.

This method never returns Ok for non-notification messages – it loops internally, discarding ParameterStatus, NoticeResponse, etc.

pub async fn close(self) -> Result<(), DriverError>

Send Terminate and close the connection.

pub fn is_idle(&self) -> bool

Whether the connection is in an idle transaction state.

pub fn is_in_transaction(&self) -> bool

Whether the connection is in a transaction.

pub fn is_in_failed_transaction(&self) -> bool

Whether the connection is in a failed transaction.

pub fn touch(&mut self)

Record that the connection was just used. Called after successful query completion so the pool can detect stale connections.

pub fn idle_duration(&self) -> Duration

How long since this connection last completed a query.

pub fn parameter(&self, name: &str) -> Option<&str>

Get a server parameter value (set during startup or via SET).

pub fn server_params(&self) -> &[(Box<str>, Box<str>)]

All server parameters received during startup.

pub fn pid(&self) -> i32

Backend process ID (for cancel requests).

pub fn secret_key(&self) -> i32

Backend secret key (for cancel requests).

pub async fn cancel(&self, config: &Config) -> Result<(), DriverError>

Cancel the currently running query on this connection.

Opens a NEW TCP connection to the same host:port and sends a CancelRequest message (16 bytes: length=16, code=80877102, pid, secret). The cancel connection is closed immediately after sending.

The config is needed to get the host:port for the new TCP connection.

pub fn is_streaming(&self) -> bool

Whether a streaming query is in progress.

pub fn drain_notifications(&mut self) -> Vec<Notification>

Drain all buffered notifications received during query processing.

Returns the pending notifications and clears the buffer. Notifications arrive asynchronously from PG (via LISTEN/NOTIFY) and are buffered during normal query execution instead of being dropped.

pub fn pending_notification_count(&self) -> usize

Number of pending notifications in the buffer.

pub fn set_max_stmt_cache_size(&mut self, size: usize)

Set the maximum number of cached prepared statements.

When the cache exceeds this size, the least recently used statement is evicted and a Close message is sent to PG to free server memory. Default: 256.

pub fn stmt_cache_len(&self) -> usize

Number of currently cached prepared statements.

pub fn created_at(&self) -> Instant

When this connection was created.