Struct BaseHandler 

pub struct BaseHandler { /* private fields */ }

Core query executor: transactional writes via sqlx, and optionally analytical reads via a pluggable ReadEngine (DuckDB or DataFusion).

Implementations

impl BaseHandler

pub fn new(pool: AnyPool) -> Self

Create a handler for writes against the given sqlx pool.

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

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

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

Create a handler with DuckDB and a live Postgres attachment.

DuckDB queries the Postgres tables directly via the pg catalog (SELECT … FROM pg.<schema>.<table>) without an explicit sync — the pre-rewrite zero-copy ATTACH pipeline. You can still also sync_* tables into local memory for faster repeated analytics.

pub fn with_datafusion(pool: AnyPool) -> Result<Self, DbkitError>

Create a handler with a DataFusion analytical read engine.

pub fn has_read_engine(&self) -> bool

Whether an analytical read engine is attached.

pub fn pool(&self) -> &AnyPool

Get a reference to the 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.

NFD alone only equalizes composed vs decomposed representations of the SAME accented string; the combining marks survive, so an accented name never matched its accent-stripped variant (a common shape across data feeds — US sources routinely strip diacritics). Stripping the marks after decomposition makes the comparison genuinely accent-insensitive.

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

Execute a write operation against the transactional pool.

Placeholders are backend-native: $1, $2, … for Postgres, ? for MySQL/SQLite. sqlx’s Any driver does not rewrite them, so write the SQL for the backend you connected to.

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

Execute an analytical query against the attached read engine, returning columnar Arrow RecordBatches.

Returns DbkitError::NoReadEngine if no engine is attached.

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

where T: DeserializeOwned,

Execute an analytical query and deserialize each row into T.

This is the typed-read replacement for the old closure-mapped ReadOp::Standard: instead of a |row| … closure, derive serde::Deserialize on your row struct. Works for any read engine, since it deserializes from the Arrow batches via serde_arrow.

#[derive(serde::Deserialize)]
struct Item { name: String, qty: i64 }
let items: Vec<Item> = handler.execute_read_as("SELECT name, qty FROM items", &[]).await?;

pub async fn sync_query( &self, name: &str, query: &str, params: &[DbValue], ) -> Result<(), DbkitError>

Run a query against the transactional pool and load its result into the analytical engine as a named in-memory table.

This is the engine-agnostic replacement for the old DuckDB ATTACH sync: rows are fetched over sqlx, converted to Arrow, and handed to the active read engine. Works for any backend × engine combination.

An empty result drops the analytical table (the schema can’t be inferred from zero rows): reads of a table synced empty error with “table not found” rather than silently serving rows from a previous sync.

pub async fn sync_tables(&self, tables: &[&str]) -> Result<(), DbkitError>

Copy entire tables from the transactional store into the analytical engine, one table per name (SELECT * FROM {table}).