oxisql-postgres — Pure-Rust PostgreSQL backend for OxiSQL

PostgreSQL backend over the tokio-postgres wire protocol with OxiTLS / RustCrypto TLS (no libpq, no openssl-sys); COPY, LISTEN/NOTIFY, pipeline batching, and extended type mapping.

Status: Stable.

What it is

oxisql-postgres provides PgConnection, which implements oxisql_core::Connection over tokio-postgres — speaking the PostgreSQL Frontend/Backend Protocol version 3 directly on the wire. There is no C client library: no libpq, no libssl, no openssl-sys, and no ring. TLS is routed entirely through OxiTLS and the rustls + rustls-rustcrypto RustCrypto provider, so a build of this crate is 100% Pure Rust and needs no system C toolchain or *-dev headers.

Beyond plain execute/query, the crate exposes the higher-value parts of the PostgreSQL protocol that thin wrappers usually omit: the COPY bulk load/unload protocol, LISTEN/NOTIFY asynchronous notifications delivered as a Stream, pipeline batching (many statements flushed in a single network round-trip), prepared-statement caching, and an extended type mapping that round-trips all 13 oxisql_core::Value variants (UUID, JSONB, NUMERIC, arrays, timestamps, …).

It is part of the OxiSQL Pure-Rust workspace and is the backend selected when you open a postgres:// URL through the oxisql facade.

Installation

[dependencies]
oxisql-postgres = "0.3.3"
# oxisql-postgres = { version = "0.3.3", features = ["replication"] }  # PostgreSQL logical replication (pgoutput)

MSRV 1.89 · edition 2021 · Apache-2.0.

Quick start

use oxisql_postgres::{PgConnection, TlsMode};
use oxisql_core::Connection;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Plain TCP (suitable for localhost / trusted networks).
    let conn = PgConnection::connect(
        "host=localhost port=5432 user=postgres password=secret dbname=mydb",
        TlsMode::Disabled,
    ).await?;

    conn.execute("CREATE TABLE IF NOT EXISTS t (id BIGINT, val TEXT)", &[]).await?;
    conn.execute("INSERT INTO t VALUES ($1, $2)", &[&1i64, &"hello"]).await?;

    let rows = conn.query("SELECT id, val FROM t WHERE id = $1", &[&1i64]).await?;
    let id: i64 = rows[0].try_get("id")?;
    println!("id = {id}");
    Ok(())
}

TLS via OxiTLS (RustCrypto provider)

use oxisql_postgres::{PgConnection, TlsMode};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Trust the Mozilla CA bundle, built with the pure-Rust RustCrypto provider.
    let root_store = oxitls::webpki_root_certs();
    let client_cfg = oxitls::client_config(root_store)
        .map_err(|e| format!("TLS config: {e}"))?;

    let conn = PgConnection::connect(
        "host=db.example.com port=5432 user=postgres dbname=mydb sslmode=require",
        TlsMode::Rustls(client_cfg),
    ).await?;
    let _ = conn;
    Ok(())
}

Two ready-made TlsMode constructors cover common cases without hand-building a ClientConfig: TlsMode::skip_verify() (accept any server certificate — dev only) and TlsMode::with_ca_pem(pem_bytes) (trust a custom CA from PEM). Both default to the RustCrypto crypto provider.

Fluent builder

use oxisql_postgres::{PgConnectionBuilder, TlsMode};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let conn = PgConnectionBuilder::new()
        .host("localhost")
        .port(5432)
        .user("postgres")
        .password("secret")
        .dbname("mydb")
        .connect_timeout_secs(5)
        .tls_mode(TlsMode::Disabled)
        .connect()
        .await?;
    let _ = conn;
    Ok(())
}

Key API

Connection timeouts

PgConnection::connect itself applies no timeout — connecting to an unreachable host can hang. PgConnection::connect_with_timeout(conn_str, tls, duration) wraps the same connect path in a tokio::time::timeout, and as of 0.3.3 it is also what the oxisql facade calls into: oxisql::connect() / connect_with_options / connect_with_tls now go through connect_with_timeout with a 10-second default (overridable via the facade's ConnectOptions::connect_timeout_ms), so a connection attempt that used to hang indefinitely now fails with a typed timeout error instead. The fix lives here, in this crate's connect_with_timeout; what's new in 0.3.3 is that the facade's plain connect() applies it automatically rather than leaving the connection unbounded.

Feature flags

Capabilities

COPY — bulk load / unload (src/copy.rs)

# async fn run(conn: &oxisql_postgres::PgConnection) -> Result<(), Box<dyn std::error::Error>> {
// Bulk-insert via COPY ... FROM STDIN (text/TSV).
let rows = vec![
    vec!["1".to_string(), "alice".to_string()],
    vec!["2".to_string(), "bob".to_string()],
];
let n = conn.copy_in_text("users", &["id", "name"], rows.into_iter()).await?;

// Extract rows via COPY ... TO STDOUT.
let out: Vec<Vec<String>> = conn.copy_out_text("users", &["id", "name"]).await?;
let _ = (n, out);
# Ok(())
# }

LISTEN / NOTIFY (src/notify.rs)

# async fn run(conn: &oxisql_postgres::PgConnection) -> Result<(), Box<dyn std::error::Error>> {
use std::time::Duration;

let mut stream = conn.listen("my_channel").await?;   // NotificationStream
conn.notify("my_channel", "ping").await?;
if let Some(n) = stream.recv_timeout(Duration::from_secs(1)).await {
    println!("{} -> {}", n.channel, n.payload);       // PgNotification
}
stream.unlisten().await?;
# Ok(())
# }

Notifications are only routed on connections created via PgConnection::connect (the background driver that dispatches NotificationResponse is spawned there); they are not available on a from_client connection.

Pipeline batching (src/pipeline.rs)

# async fn run(conn: &oxisql_postgres::PgConnection) -> Result<(), Box<dyn std::error::Error>> {
let mut pipeline = conn.pipeline();
pipeline.add_execute("INSERT INTO t VALUES ($1)", &[&1i64]);
pipeline.add_query("SELECT * FROM t", &[]);
let result = pipeline.finish().await?;   // PipelineResult, single round-trip
let _ = result;
# Ok(())
# }

Prepared statements, describe & introspection

prepare(sql) returns a PgPrepared whose underlying Statement is cached by SQL hash, so re-preparing the same text reuses the server-side plan. describe(sql) returns Vec<ColumnDescription> without executing the query. Schema introspection (tables, columns, indexes, foreign_keys) is served from information_schema and pg_indexes.

Logical replication (replication feature, src/replication/)

Behind the non-default replication Cargo feature, this crate can act as a logical-replication client speaking PostgreSQL's pgoutput output-plugin wire format directly. tokio-postgres cannot negotiate CopyBoth/replication mode, so this path never goes through it: PgReplicationConnection drives the wire protocol itself (via postgres-protocol + fallible-iterator), reusing the crate's own connection-string parsing and full TLS / SCRAM-SHA-256 auth stack.

use futures::StreamExt;
use oxisql_postgres::{LogicalReplicationMessage, PgReplicationConnection, ReplicationEvent, TlsMode};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut repl = PgReplicationConnection::connect(
        "host=localhost user=postgres dbname=mydb",
        TlsMode::Disabled,
    ).await?;

    let ident = repl.identify_system().await?;
    println!("system id: {}", ident.systemid);

    let slot = repl.create_replication_slot("my_slot", false).await?;
    let mut stream = repl
        .start_logical_replication("my_slot", &["my_publication"], slot.consistent_point)
        .await?;

    while let Some(event) = stream.next().await {
        if let ReplicationEvent::Logical {
            message: LogicalReplicationMessage::Commit { commit_lsn, .. },
            ..
        } = event?
        {
            stream.ack(commit_lsn).await?; // acknowledge after durably applying
        }
    }
    Ok(())
}

(Requires the replication feature; cargo add oxisql-postgres --features replication.)

• Connection setup — PgReplicationConnection::connect performs the replication=database handshake (a separate connection mode from ordinary PgConnection), then identify_system() / create_replication_slot(name, temporary) / drop_replication_slot(name) run over the simple-query protocol.
• Streaming — start_logical_replication(slot_name, publication_names, start_lsn) consumes the connection and returns a ReplicationStream (impl futures::Stream<Item = Result<ReplicationEvent, PgError>>), backed by a background reader task plus a periodic Standby Status Update keepalive task (every 10s).
• Events — ReplicationEvent::Logical carries a decoded LogicalReplicationMessage (Begin / Commit / Origin / Relation / Type / Insert / Update / Delete / Truncate / Message); ReplicationEvent::KeepAlive is a server liveness probe.
• Progress tracking — stream.ack(lsn) (or the more general stream.standby_status_update(written, flushed, applied)) after durably applying a transaction.
• Tuple decoding — stream.relation(rel_id) returns the cached RelationBody schema; stream.decode_tuple(rel_id, tuple) maps a TupleData to Vec<CellValue> (each cell either a decoded oxisql_core::Value or UnchangedToast, for an un-retransmitted TOASTed column). Both pgoutput wire formats are decoded — text (the format this MVP always negotiates) and binary — including PostgreSQL's array-literal text syntax ({1,2,3}, {}, {NULL,2}, quoting/escaping, multi-dimensional arrays).
• LSN helpers — Lsn wraps a WAL position and round-trips PostgreSQL's canonical text form ("16/B374D848") via FromStr/Display; pg_micros_to_unix_micros / unix_micros_to_pg_micros convert between the PostgreSQL-epoch and Unix-epoch timestamp conventions used on the wire.

This is an MVP scoped to what a CDC (change-data-capture) consumer needs first: not implemented yet are streaming of large in-progress transactions (streaming 'on'), two-phase commit, and parallel streaming (see the crate's TODO.md for the tracked follow-ups); physical replication is out of scope entirely (a different protocol from pgoutput decoding). Live-server integration tests live in tests/replication.rs, gated behind integration-postgres,replication and individually #[ignore]d pending a real wal_level=logical server.

Type mapping

All 13 oxisql_core::Value variants round-trip. Parameters are encoded with explicit type OIDs; results can be requested in binary format (query_binary).

Test coverage

68 tests passing under plain cargo test (no extra features; includes 15 doctests), plus 6 #[ignore]d live-server tests (TLS live connect, COPY IN/OUT, 4× LISTEN/NOTIFY) that are skipped without a real PostgreSQL instance.

With --all-features (adds integration-postgres + replication) the suite grows substantially: 393 tests run under cargo nextest (doctests excluded), plus 41 more that stay #[ignore]d pending a live server — the 6 above, a 30-test CRUD/isolation/reconnect/pooling/prepared-statement/type-mapping integration suite, and the 5-test tests/replication.rs logical-replication suite (needs wal_level=logical; see "Logical replication" above). The replication feature's own unit-test suite alone (cargo test -p oxisql-postgres --features replication --lib) totals 339 tests, covering pgoutput message decoding, LSN parsing, CopyBoth framing, SCRAM-SHA-256 auth, and text-/binary-format tuple and array decoding.

cargo test -p oxisql-postgres                                          # 68 passed (incl. doctests), live tests skipped
cargo nextest run -p oxisql-postgres --all-features --run-ignored all  # also runs every live-server test (needs a real server)

Part of the OxiSQL workspace

This crate is one of 17 crates in the Pure-Rust OxiSQL workspace (2,157 workspace tests pass, 2,651 with --all-features). It is the PostgreSQL backend behind the unified oxisql facade; see the workspace README for the embedded engine, MySQL backend, connection pool, migrations, and DataFusion integration.

License

Apache-2.0 — Copyright © 2024–2026 COOLJAPAN OU (Team Kitasan).