Struct PgDriver 

pub struct PgDriver { /* private fields */ }

Combines the pure encoder (Layer 2) with async I/O (Layer 3).

Implementations

impl PgDriver

pub fn new(connection: PgConnection) -> Self

Create a new driver with an existing connection.

pub fn builder() -> PgDriverBuilder

Builder pattern for ergonomic connection configuration.

Example

let driver = PgDriver::builder()
    .host("localhost")
    .port(5432)
    .user("admin")
    .database("mydb")
    .password("secret")  // Optional
    .connect()
    .await?;

pub async fn connect( host: &str, port: u16, user: &str, database: &str, ) -> PgResult<Self>

Connect to PostgreSQL and create a driver (trust mode, no password).

Arguments

• host — PostgreSQL server hostname or IP.
• port — TCP port (typically 5432).
• user — PostgreSQL role name.
• database — Target database name.

pub async fn connect_with_password( host: &str, port: u16, user: &str, database: &str, password: &str, ) -> PgResult<Self>

Connect to PostgreSQL with password authentication. Supports server-requested auth flow: cleartext, MD5, or SCRAM-SHA-256.

pub async fn connect_with_options( host: &str, port: u16, user: &str, database: &str, password: Option<&str>, options: ConnectOptions, ) -> PgResult<Self>

Connect with explicit security options.

pub async fn connect_logical_replication( host: &str, port: u16, user: &str, database: &str, password: Option<&str>, ) -> PgResult<Self>

Connect in logical replication mode (replication=database).

This enables replication commands such as IDENTIFY_SYSTEM and CREATE_REPLICATION_SLOT.

pub async fn connect_logical_replication_with_options( host: &str, port: u16, user: &str, database: &str, password: Option<&str>, options: ConnectOptions, ) -> PgResult<Self>

Connect with explicit options and force logical replication mode.

pub async fn connect_env() -> PgResult<Self>

Connect using DATABASE_URL environment variable.

Parses the URL format: postgresql://user:password@host:port/database or postgres://user:password@host:port/database

Example

// Set DATABASE_URL=postgresql://user:pass@localhost:5432/mydb
let driver = PgDriver::connect_env().await?;

pub async fn connect_url(url: &str) -> PgResult<Self>

Connect using a PostgreSQL connection URL.

Parses the URL format: postgresql://user:password@host:port/database?params or postgres://user:password@host:port/database?params

Supports all enterprise query params (sslmode, auth_mode, gss_provider, channel_binding, etc.) — same set as PoolConfig::from_qail_config.

Example

let driver = PgDriver::connect_url("postgresql://user:pass@localhost:5432/mydb?sslmode=require").await?;

pub async fn connect_with_timeout( host: &str, port: u16, user: &str, database: &str, password: &str, timeout: Duration, ) -> PgResult<Self>

Connect to PostgreSQL with a connection timeout. If the connection cannot be established within the timeout, returns an error.

Example

use std::time::Duration;
let driver = PgDriver::connect_with_timeout(
    "localhost", 5432, "user", "db", "password",
    Duration::from_secs(5)
).await?;

pub fn clear_cache(&mut self)

Clear the prepared statement cache. Frees memory by removing all cached statements. Note: Statements remain on the PostgreSQL server until connection closes.

pub fn cache_stats(&self) -> (usize, usize)

Get cache statistics. Returns (current_size, max_capacity).

impl PgDriver

pub async fn fetch_all(&mut self, cmd: &Qail) -> PgResult<Vec<PgRow>>

Execute a QAIL command and fetch all rows (CACHED + ZERO-ALLOC). Default method - uses prepared statement caching for best performance. On first call: sends Parse + Bind + Execute + Sync On subsequent calls with same SQL: sends only Bind + Execute (SKIPS Parse!) Uses LRU cache with max 1000 statements (auto-evicts oldest).

pub async fn fetch_all_with_format( &mut self, cmd: &Qail, result_format: ResultFormat, ) -> PgResult<Vec<PgRow>>

Execute a QAIL command and fetch all rows using a specific result format.

result_format controls server result-column encoding:

• ResultFormat::Text for standard text decoding.
• ResultFormat::Binary for binary wire values.

pub async fn fetch_typed<T: QailRow>(&mut self, cmd: &Qail) -> PgResult<Vec<T>>

Execute a QAIL command and fetch all rows as a typed struct (text format). Requires the target type to implement QailRow trait.

Example

let users: Vec<User> = driver.fetch_typed::<User>(&query).await?;

pub async fn fetch_typed_with_format<T: QailRow>( &mut self, cmd: &Qail, result_format: ResultFormat, ) -> PgResult<Vec<T>>

Execute a QAIL command and fetch all rows as a typed struct with explicit result format.

Use ResultFormat::Binary to get binary wire values; row decoding should use metadata-aware accessors such as PgRow::try_get() / try_get_by_name().

pub async fn fetch_one_typed<T: QailRow>( &mut self, cmd: &Qail, ) -> PgResult<Option<T>>

Execute a QAIL command and fetch a single row as a typed struct (text format). Returns None if no rows are returned.

pub async fn fetch_one_typed_with_format<T: QailRow>( &mut self, cmd: &Qail, result_format: ResultFormat, ) -> PgResult<Option<T>>

Execute a QAIL command and fetch a single row as a typed struct with explicit result format.

pub async fn fetch_all_uncached(&mut self, cmd: &Qail) -> PgResult<Vec<PgRow>>

Execute a QAIL command and fetch all rows (UNCACHED). Sends Parse + Bind + Execute on every call. Use for one-off queries or when caching is not desired.

Optimized: encodes wire bytes into reusable write_buf (zero-alloc).

pub async fn fetch_all_uncached_with_format( &mut self, cmd: &Qail, result_format: ResultFormat, ) -> PgResult<Vec<PgRow>>

Execute a QAIL command and fetch all rows (UNCACHED) with explicit result format.

pub async fn fetch_all_fast(&mut self, cmd: &Qail) -> PgResult<Vec<PgRow>>

Execute a QAIL command and fetch all rows (FAST VERSION). Uses optimized recv_with_data_fast for faster response parsing. Skips column metadata collection for maximum speed.

pub async fn fetch_all_fast_with_format( &mut self, cmd: &Qail, result_format: ResultFormat, ) -> PgResult<Vec<PgRow>>

Execute a QAIL command and fetch all rows (FAST VERSION) with explicit result format.

pub async fn fetch_one(&mut self, cmd: &Qail) -> PgResult<PgRow>

Execute a QAIL command and fetch one row.

pub async fn fetch_all_cached(&mut self, cmd: &Qail) -> PgResult<Vec<PgRow>>

Execute a QAIL command with PREPARED STATEMENT CACHING. Like fetch_all(), but caches the prepared statement on the server. On first call: sends Parse + Describe + Bind + Execute + Sync On subsequent calls: sends only Bind + Execute + Sync (SKIPS Parse!) Column metadata (RowDescription) is cached alongside the statement so that by-name column access works on every call.

Optimized: all wire messages are batched into a single write_all syscall.

pub async fn fetch_all_cached_with_format( &mut self, cmd: &Qail, result_format: ResultFormat, ) -> PgResult<Vec<PgRow>>

Execute a QAIL command with prepared statement caching and explicit result format.

pub async fn execute(&mut self, cmd: &Qail) -> PgResult<u64>

Execute a QAIL command (for mutations) - ZERO-ALLOC.

pub async fn query_ast(&mut self, cmd: &Qail) -> PgResult<QueryResult>

Query a QAIL command and return rows (for SELECT/GET queries). Like execute() but collects RowDescription + DataRow messages instead of discarding them.

pub async fn query_ast_with_format( &mut self, cmd: &Qail, result_format: ResultFormat, ) -> PgResult<QueryResult>

Query a QAIL command and return rows using an explicit result format.

impl PgDriver

pub async fn begin(&mut self) -> PgResult<()>

Begin a transaction (AST-native).

pub async fn commit(&mut self) -> PgResult<()>

Commit the current transaction (AST-native).

pub async fn rollback(&mut self) -> PgResult<()>

Rollback the current transaction (AST-native).

pub async fn savepoint(&mut self, name: &str) -> PgResult<()>

Create a named savepoint within the current transaction. Savepoints allow partial rollback within a transaction. Use rollback_to() to return to this savepoint.

Example

driver.begin().await?;
driver.execute(&insert1).await?;
driver.savepoint("sp1").await?;
driver.execute(&insert2).await?;
driver.rollback_to("sp1").await?; // Undo insert2, keep insert1
driver.commit().await?;

pub async fn rollback_to(&mut self, name: &str) -> PgResult<()>

Rollback to a previously created savepoint. Discards all changes since the named savepoint was created, but keeps the transaction open.

pub async fn release_savepoint(&mut self, name: &str) -> PgResult<()>

Release a savepoint (free resources, if no longer needed). After release, the savepoint cannot be rolled back to.

pub async fn execute_batch(&mut self, cmds: &[Qail]) -> PgResult<Vec<u64>>

Execute multiple commands in a single atomic transaction. All commands succeed or all are rolled back.

Example

let cmds = vec![
    Qail::add("users").columns(["name"]).values(["Alice"]),
    Qail::add("users").columns(["name"]).values(["Bob"]),
];
let results = driver.execute_batch(&cmds).await?;
// results = [1, 1] (rows affected)

pub async fn set_statement_timeout(&mut self, ms: u32) -> PgResult<()>

Set statement timeout for this connection (in milliseconds).

Example

driver.set_statement_timeout(30_000).await?; // 30 seconds

pub async fn reset_statement_timeout(&mut self) -> PgResult<()>

Reset statement timeout to default (no limit).

pub async fn set_rls_context(&mut self, ctx: RlsContext) -> PgResult<()>

Set the RLS context for multi-tenant data isolation.

Configures PostgreSQL session variables (app.current_tenant_id, etc.) so that RLS policies automatically filter data by tenant.

Since PgDriver takes &mut self, the borrow checker guarantees that set_config and all subsequent queries execute on the same connection — no pool race conditions possible.

Example

driver.set_rls_context(RlsContext::tenant("tenant-123")).await?;
let orders = driver.fetch_all(&Qail::get("orders")).await?;
// orders only contains rows for tenant-123

pub async fn clear_rls_context(&mut self) -> PgResult<()>

Clear the RLS context, resetting session variables to safe defaults.

After clearing, all RLS-protected queries will return zero rows (empty tenant scope matches nothing).

pub fn rls_context(&self) -> Option<&RlsContext>

Get the current RLS context, if any.

pub async fn pipeline_batch(&mut self, cmds: &[Qail]) -> PgResult<usize>

Execute multiple Qail ASTs in a single network round-trip (PIPELINING).

Example

let cmds: Vec<Qail> = (1..=1000)
    .map(|i| Qail::get("harbors").columns(["id", "name"]).limit(i))
    .collect();
let count = driver.pipeline_batch(&cmds).await?;
assert_eq!(count, 1000);

pub async fn pipeline_fetch( &mut self, cmds: &[Qail], ) -> PgResult<Vec<Vec<PgRow>>>

Execute multiple Qail ASTs and return full row data.

pub async fn prepare(&mut self, sql: &str) -> PgResult<PreparedStatement>

Prepare a SQL statement for repeated execution.

pub async fn pipeline_prepared_fast( &mut self, stmt: &PreparedStatement, params_batch: &[Vec<Option<Vec<u8>>>], ) -> PgResult<usize>

Execute a prepared statement pipeline in FAST mode (count only).

pub async fn copy_bulk( &mut self, cmd: &Qail, rows: &[Vec<Value>], ) -> PgResult<u64>

Bulk insert data using PostgreSQL COPY protocol (AST-native). Uses a Qail::Add to get validated table and column names from the AST, not user-provided strings. This is the sound, AST-native approach.

Example

// Create a Qail::Add to define table and columns
let cmd = Qail::add("users")
    .columns(["id", "name", "email"]);
// Bulk insert rows
let rows: Vec<Vec<Value>> = vec![
    vec![Value::Int(1), Value::String("Alice"), Value::String("alice@ex.com")],
    vec![Value::Int(2), Value::String("Bob"), Value::String("bob@ex.com")],
];
driver.copy_bulk(&cmd, &rows).await?;

pub async fn copy_bulk_bytes( &mut self, cmd: &Qail, data: &[u8], ) -> PgResult<u64>

Fastest bulk insert using pre-encoded COPY data. Accepts raw COPY text format bytes. Use when caller has already encoded rows to avoid any encoding overhead.

Format

Data should be tab-separated rows with newlines (COPY text format): 1\thello\t3.14\n2\tworld\t2.71\n

Example

let cmd = Qail::add("users").columns(["id", "name"]);
let data = b"1\tAlice\n2\tBob\n";
driver.copy_bulk_bytes(&cmd, data).await?;

pub async fn copy_export_table( &mut self, table: &str, columns: &[String], ) -> PgResult<Vec<u8>>

Export table data using PostgreSQL COPY TO STDOUT (zero-copy streaming). Returns rows as tab-separated bytes for direct re-import via copy_bulk_bytes.

Example

let data = driver.copy_export_table("users", &["id", "name"]).await?;
shadow_driver.copy_bulk_bytes(&cmd, &data).await?;

pub async fn copy_export_table_stream<F, Fut>( &mut self, table: &str, columns: &[String], on_chunk: F, ) -> PgResult<()>

where F: FnMut(Vec<u8>) -> Fut, Fut: Future<Output = PgResult<()>>,

Stream table export using COPY TO STDOUT with bounded memory usage.

Chunks are forwarded directly from PostgreSQL to on_chunk.

pub async fn copy_export_cmd_stream<F, Fut>( &mut self, cmd: &Qail, on_chunk: F, ) -> PgResult<()>

where F: FnMut(Vec<u8>) -> Fut, Fut: Future<Output = PgResult<()>>,

Stream an AST-native Qail::Export command as raw COPY chunks.

pub async fn copy_export_cmd_stream_rows<F>( &mut self, cmd: &Qail, on_row: F, ) -> PgResult<()>

where F: FnMut(Vec<String>) -> PgResult<()>,

Stream an AST-native Qail::Export command as parsed text rows.

pub async fn stream_cmd( &mut self, cmd: &Qail, batch_size: usize, ) -> PgResult<Vec<Vec<PgRow>>>

Stream large result sets using PostgreSQL cursors. This method uses DECLARE CURSOR internally to stream rows in batches, avoiding loading the entire result set into memory.

Example

let cmd = Qail::get("large_table");
let batches = driver.stream_cmd(&cmd, 100).await?;
for batch in batches {
    for row in batch {
        // process row
    }
}

impl PgDriver

pub async fn identify_system(&mut self) -> PgResult<IdentifySystem>

Driver wrapper for PgConnection::identify_system.

pub async fn create_logical_replication_slot( &mut self, slot_name: &str, output_plugin: &str, temporary: bool, two_phase: bool, ) -> PgResult<ReplicationSlotInfo>

Driver wrapper for PgConnection::create_logical_replication_slot.

pub async fn drop_replication_slot( &mut self, slot_name: &str, wait: bool, ) -> PgResult<()>

Driver wrapper for PgConnection::drop_replication_slot.

pub async fn start_logical_replication( &mut self, slot_name: &str, start_lsn: &str, options: &[ReplicationOption], ) -> PgResult<ReplicationStreamStart>

Driver wrapper for PgConnection::start_logical_replication.

pub async fn recv_replication_message( &mut self, ) -> PgResult<ReplicationStreamMessage>

Driver wrapper for PgConnection::recv_replication_message.

pub async fn send_standby_status_update( &mut self, write_lsn: u64, flush_lsn: u64, apply_lsn: u64, reply_requested: bool, ) -> PgResult<()>

Driver wrapper for PgConnection::send_standby_status_update.