Struct PgHandler 

pub struct PgHandler { /* private fields */ }

Core query executor for native Postgres: rich-typed transactional writes via sqlx, and row-mapped analytical reads via DuckDB.

Implementations

impl PgHandler

pub fn new(pool: PgPool) -> Self

Create a handler for writes against the given native Postgres pool.

pub fn with_duckdb(pool: PgPool) -> Result<Self, DbkitError>

Create a handler with an in-memory DuckDB analytical read engine.

pub fn with_duckdb_attached_postgres( pool: PgPool, pg_connection_string: &str, ) -> Result<Self, DbkitError>

Create a handler with DuckDB and a live Postgres attachment, so DuckDB queries the Postgres tables directly via the pg catalog (SELECT … FROM pg.<schema>.<table>) without an explicit sync.

pub fn has_read_engine(&self) -> bool

Whether a DuckDB read engine is attached.

pub fn pool(&self) -> &PgPool

Get a reference to the native Postgres write pool.

pub fn normalize_name(name: &str) -> String

Accent-insensitive name key: NFD-decompose, DROP combining marks, then lowercase — “José Ramírez” and “Jose Ramirez” produce the same key. See BaseHandler::normalize_name for the rationale (NFD alone leaves combining marks, so accented names never matched stripped variants).

pub async fn execute_write( &self, op: WriteOp<'_>, ) -> Result<QueryResult<PgRow>, DbkitError>

Execute a write operation against the Postgres pool. Placeholders are Postgres-native ($1, $2, …).

pub async fn copy_in( &self, table: &str, columns: &[&str], rows: &[Vec<DbValue>], ) -> Result<u64, DbkitError>

Bulk-insert rows via Postgres COPY ... FROM STDIN (text format).

The fastest way to load many rows — one streamed COPY instead of a parse + execute (+ savepoint) per row like WriteOp::BatchParams. Benchmarks at roughly 30–50× the throughput of BatchParams. Each row in rows must align positionally with columns. Returns the number of rows copied.

copy_in vs WriteOp::BatchParams — which to use

Reach for copy_in when…	Reach for BatchParams when…
Plain bulk insert into one table	You need INSERT … ON CONFLICT (upsert)
Data is trusted; all-or-nothing is fine	You need per-row isolation (skip bad rows)
You want maximum throughput	The statement isn’t a plain insert (UPDATE, RETURNING, computed VALUES)
Target is Postgres	Target is a non-Postgres backend (use the Any pool)

COPY is not an INSERT statement, so it does not support ON CONFLICT, RETURNING, DEFAULT expressions, or WHERE, and it is all-or-nothing: a constraint violation aborts the entire load (it does not skip bad rows like BatchParams).

To bulk-upsert, combine the two: COPY into a constraint-free staging table, then run one set-based INSERT … SELECT … ON CONFLICT — far faster than per-row BatchParams with ON CONFLICT:

CREATE TEMP TABLE stage (LIKE target INCLUDING DEFAULTS) ON COMMIT DROP;
COPY stage (id, name) FROM STDIN;            -- fast bulk load, no constraints
INSERT INTO target (id, name)
  SELECT id, name FROM stage                 -- one set-based upsert
  ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name;

pub async fn copy_upsert( &self, table: &str, columns: &[&str], conflict_columns: &[&str], update_columns: &[&str], rows: &[Vec<DbValue>], ) -> Result<u64, DbkitError>

Bulk-upsert rows: COPY into a staging table, then one set-based INSERT … SELECT … ON CONFLICT, all in a single transaction.

This is the fast path for ON CONFLICT at scale. Plain copy_in can’t do ON CONFLICT (it’s not an INSERT), and per-row WriteOp::BatchParams with ON CONFLICT pays per-row overhead. This combines both strengths: COPY’s bulk ingestion into a constraint-free staging table, then a single set-based upsert into the target.

• columns — columns present in rows (positional), copied into staging.
• conflict_columns — the conflict target (must back a unique/PK index).
• update_columns — columns to overwrite on conflict (set to the incoming EXCLUDED value). Empty ⇒ DO NOTHING (insert-or-ignore).

Returns the number of rows inserted or updated.

The staging table is CREATE TEMP TABLE … AS SELECT {columns} FROM target WITH NO DATA (ON COMMIT DROP) — target column types, no constraints or defaults — and vanishes at commit. The final upsert is all-or-nothing: a non-conflict error (CHECK/FK/type) aborts the batch. Within a single call, conflict_columns must be unique across rows — duplicate keys make ON CONFLICT DO UPDATE error with “command cannot affect row a second time”; de-duplicate before calling.

pub async fn query_scalar_opt<T>( &self, query: &str, params: Vec<DbValue>, ) -> Result<Option<T>, DbkitError>

where T: for<'r> Decode<'r, Postgres> + Type<Postgres>,

Run a single-value SELECT and return the first column of the single matching row, or None if no row matched.

Convenience over the execute_write(WriteOp::Single { .., FetchMode::Optional }) → .optional()? → row.get(0) dance that OLTP get_*_id / get_or_create lookups hand-roll everywhere. Uses Optional (not One), so an empty result is Ok(None) rather than an error; a genuine DB failure still surfaces as Err.

pub async fn query_scalar<T>( &self, query: &str, params: Vec<DbValue>, ) -> Result<T, DbkitError>

where T: for<'r> Decode<'r, Postgres> + Type<Postgres>,

Run a single-value SELECT (or INSERT … RETURNING) that must yield exactly one row, and return the first column. Errors if no row matched — use query_scalar_opt when zero rows is valid.

pub async fn query_col<T>( &self, query: &str, params: Vec<DbValue>, ) -> Result<Vec<T>, DbkitError>

where T: for<'r> Decode<'r, Postgres> + Type<Postgres>,

Run a SELECT and collect the first column of every row into a Vec.

pub async fn query( &self, query: &str, params: Vec<DbValue>, mode: FetchMode, ) -> Result<QueryResult<PgRow>, DbkitError>

Run a query against the native Postgres pool, returning rows per mode.

This is the OLTP read path — single-row lookups and small result sets go straight to Postgres (one round-trip → PgRow), no analytical engine. Placeholders are Postgres-native ($1, $2, …); read columns off the returned PgRows with row.get(i) / row.try_get(i).

pub async fn execute_read( &self, sql: &str, params: &[DbValue], ) -> Result<Vec<RecordBatch>, DbkitError>

Run an analytical query against the attached DuckDB engine, returning columnar Arrow RecordBatches. For large joins/aggregations consumed as DataFrames. Errors with DbkitError::NoReadEngine if no engine is attached. For typed rows, deserialize the batches (see BaseHandler::execute_read_as).

pub async fn execute_read_as<T>( &self, sql: &str, params: &[DbValue], ) -> Result<Vec<T>, DbkitError>

where T: DeserializeOwned,

Like execute_read but deserializes each row into T via serde_arrow — the typed analytical read. T’s field names must match the query’s output column names. Use for DuckDB-side analytical reads (large scans / aggregations) that map to typed rows. Errors with DbkitError::NoReadEngine if no engine is attached.