Struct PostgresAdapter 

pub struct PostgresAdapter { /* private fields */ }

The PostgreSQL backend, behind the same feature that gates it in fraiseql-db.

PostgresAdapter belongs here, not in the unconditional list above. The blanket pub use fraiseql_db as db; carried each item’s own #[cfg] for free; enumerating the names drops that, so every line above is an assertion that the item is ungated in fraiseql-db — and nothing but a feature-OFF build checks it. make lint-feature-matrix is that build. PostgreSQL database adapter with connection pooling.

Uses deadpool-postgres for connection pooling and tokio-postgres for async queries.

Example

use fraiseql_db::postgres::PostgresAdapter;
use fraiseql_db::{DatabaseAdapter, WhereClause, WhereOperator};
use serde_json::json;

// Create adapter with connection string
let adapter = PostgresAdapter::new("postgresql://localhost/mydb").await?;

// Execute query
let where_clause = WhereClause::Field {
    path: vec!["email".to_string()],
    operator: WhereOperator::Icontains,
    value: json!("example.com"),
};

let results = adapter
    .execute_where_query("v_user", Some(&where_clause), Some(10), None, None)
    .await?;

println!("Found {} users", results.len());

Implementations

impl PostgresAdapter

pub async fn new( connection_string: &str, ) -> Result<PostgresAdapter, FraiseQLError>

Create new PostgreSQL adapter with default pool configuration.

Arguments

• connection_string - PostgreSQL connection string (e.g., “postgresql://localhost/mydb”)

Errors

Returns FraiseQLError::ConnectionPool if pool creation fails.

Example

let adapter = PostgresAdapter::new("postgresql://localhost/mydb").await?;

pub async fn with_pool_config( connection_string: &str, cfg: PoolPrewarmConfig, ) -> Result<PostgresAdapter, FraiseQLError>

Create new PostgreSQL adapter with pre-warming and timeout configuration.

Constructs the pool, runs a startup health check, then eagerly opens cfg.min_size connections so they are ready when the first request arrives.

Arguments

• connection_string - PostgreSQL connection string
• cfg - Pool pre-warming and timeout configuration

Errors

Returns FraiseQLError::ConnectionPool if pool creation or the startup health check fails.

pub async fn with_pool_size( connection_string: &str, max_size: usize, ) -> Result<PostgresAdapter, FraiseQLError>

Create new PostgreSQL adapter with custom pool size.

Arguments

• connection_string - PostgreSQL connection string
• max_size - Maximum number of connections in pool

Errors

Returns FraiseQLError::ConnectionPool if pool creation fails.

pub const fn pool(&self) -> &Pool<Manager>

Get a reference to the internal connection pool.

This allows sharing the pool with other components like PostgresIntrospector.

pub fn with_mutation_timing(self, variable_name: &str) -> PostgresAdapter

Enable mutation timing injection.

When enabled, execute_write sets a session variable to clock_timestamp()::text in each write’s transaction before the call, allowing SQL functions to compute their own duration.

Arguments

• variable_name - The PostgreSQL session variable name (e.g., "fraiseql.started_at")

pub const fn mutation_timing_enabled(&self) -> bool

Returns whether mutation timing injection is enabled.

pub async fn execute_with_projection( &self, view: &str, projection: Option<&SqlProjectionHint>, where_clause: Option<&WhereClause>, limit: Option<u32>, offset: Option<u32>, ) -> Result<Vec<JsonbValue>, FraiseQLError>

Execute query with SQL field projection optimization.

Convenience wrapper for callers that don’t need ORDER BY. See execute_with_projection_impl for details.

Errors

Returns FraiseQLError::Database on query execution failure.

Trait Implementations

impl Clone for PostgresAdapter

fn clone(&self) -> PostgresAdapter

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl DatabaseAdapter for PostgresAdapter

fn supports_composed_reads(&self) -> bool

execute_composed_with_session is implemented below.

fn execute_raw_query<'life0, 'life1, 'async_trait>( &'life0 self, sql: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Vec<HashMap<String, Value>>, FraiseQLError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresAdapter: 'async_trait,

Security

sql must be compiler-generated. Never pass user-supplied strings directly — doing so would open SQL-injection vulnerabilities.

fn count_where_query<'life0, 'life1, 'life2, 'life3, 'life4, 'life5, 'async_trait>( &'life0 self, view: &'life1 str, where_clause: Option<&'life2 WhereClause>, session_vars: &'life3 [(&'life4 str, &'life5 str)], routing: ReadRouting, ) -> Pin<Box<dyn Future<Output = Result<u64, FraiseQLError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, 'life4: 'async_trait, 'life5: 'async_trait, PostgresAdapter: 'async_trait,

SELECT COUNT(*), pushed into the database rather than counted in Rust.

The trait’s default fetches every matching row and takes .len(), which is correct but materialises the whole filtered set — on an unfiltered list that is the entire table in memory to answer a single integer.

fn execute_composed_with_session<'life0, 'life1, 'life2, 'life3, 'life4, 'async_trait>( &'life0 self, read: &'life1 ComposedLevel, session_vars: &'life2 [(&'life3 str, &'life4 str)], routing: ReadRouting, ) -> Pin<Box<dyn Future<Output = Result<Arc<Vec<JsonbValue>>, FraiseQLError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, 'life4: 'async_trait, PostgresAdapter: 'async_trait,

One statement, one snapshot, one connection held for the delivery (#958).

The SQL is built by the same two builders the collecting read uses, so a streamed export and a buffered read of the same query cannot select different rows — the drift that made execute_query_direct disagree with count_rows about a tenant filter (#739).

fn stream_row_query<'life0, 'life1, 'life2, 'life3, 'life4, 'async_trait>( &'life0 self, view_name: &'life1 str, columns: &'life2 [ColumnSpec], where_sql: Option<&'life3 str>, order_by: Option<&'life4 str>, limit: Option<u32>, offset: Option<u32>, ) -> Pin<Box<dyn Future<Output = Result<Pin<Box<dyn Stream<Item = Result<Vec<ColumnValue>, FraiseQLError>> + Send>>, FraiseQLError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, 'life4: 'async_trait, PostgresAdapter: 'async_trait,

The column-shaped streaming read behind gRPC server-streaming (#958).

Routed to the primary like its collecting sibling: execute_row_query reaches execute_raw_query, and a surface that executes composed SQL is never sent to a replica.

fn execute_row_query_with_session<'life0, 'life1, 'life2, 'life3, 'life4, 'life5, 'life6, 'life7, 'async_trait>( &'life0 self, view_name: &'life1 str, columns: &'life2 [ColumnSpec], where_sql: Option<&'life3 str>, order_by: Option<&'life4 str>, limit: Option<u32>, offset: Option<u32>, session_vars: &'life5 [(&'life6 str, &'life7 str)], ) -> Pin<Box<dyn Future<Output = Result<Vec<Vec<ColumnValue>>, FraiseQLError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, 'life4: 'async_trait, 'life5: 'async_trait, 'life6: 'async_trait, 'life7: 'async_trait, PostgresAdapter: 'async_trait,

The session-pinned collecting row read (#1351).

Same shape as count_where_query: with no session variables it is a plain read; with them it is one transaction, so set_config and the SELECT land on the same connection and an RLS policy backed by current_setting() sees them (#329).

fn stream_row_query_with_session<'life0, 'life1, 'life2, 'life3, 'life4, 'life5, 'life6, 'life7, 'async_trait>( &'life0 self, view_name: &'life1 str, columns: &'life2 [ColumnSpec], where_sql: Option<&'life3 str>, order_by: Option<&'life4 str>, limit: Option<u32>, offset: Option<u32>, session_vars: &'life5 [(&'life6 str, &'life7 str)], ) -> Pin<Box<dyn Future<Output = Result<Pin<Box<dyn Stream<Item = Result<Vec<ColumnValue>, FraiseQLError>> + Send>>, FraiseQLError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, 'life4: 'async_trait, 'life5: 'async_trait, 'life6: 'async_trait, 'life7: 'async_trait, PostgresAdapter: 'async_trait,

The session-pinned streaming row read (#1351).

stream_rows already pins session variables for the life of the portal — the non-session arm above passes &[] to the very same call.

fn execute_admin_sql<'life0, 'life1, 'async_trait>( &'life0 self, request: &'life1 AdminSqlRequest, ) -> Pin<Box<dyn Future<Output = Result<AdminSqlOutcome, FraiseQLError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresAdapter: 'async_trait,

The operator SQL console (#962). The containment lives in the private postgres::adapter::admin_sql module; see AdminSqlRequest for what each bound is and how PostgreSQL — not a parse of the statement text — enforces it.

fn execute_with_projection<'life0, 'life1, 'life2, 'life3, 'life4, 'async_trait>( &'life0 self, view: &'life1 str, projection: Option<&'life2 SqlProjectionHint>, where_clause: Option<&'life3 WhereClause>, limit: Option<u32>, offset: Option<u32>, order_by: Option<&'life4 [OrderByClause]>, ) -> Pin<Box<dyn Future<Output = Result<Vec<JsonbValue>, FraiseQLError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, 'life4: 'async_trait, PostgresAdapter: 'async_trait,

Execute a WHERE query with SQL field projection optimization.

fn execute_where_query<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, view: &'life1 str, where_clause: Option<&'life2 WhereClause>, limit: Option<u32>, offset: Option<u32>, order_by: Option<&'life3 [OrderByClause]>, ) -> Pin<Box<dyn Future<Output = Result<Vec<JsonbValue>, FraiseQLError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, PostgresAdapter: 'async_trait,

Execute a WHERE query against a view and return JSONB rows.

fn explain_where_query<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, view: &'life1 str, where_clause: Option<&'life2 WhereClause>, limit: Option<u32>, offset: Option<u32>, ) -> Pin<Box<dyn Future<Output = Result<Value, FraiseQLError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, PostgresAdapter: 'async_trait,

Run EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) against a view with the same parameterized WHERE clause that execute_where_query would use.

fn database_type(&self) -> DatabaseType

Get database type (for logging/metrics).

fn health_check<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<(), FraiseQLError>> + Send + 'async_trait>>

where 'life0: 'async_trait, PostgresAdapter: 'async_trait,

Health check - verify database connectivity.

fn pool_metrics(&self) -> PoolMetrics

Get connection pool metrics.

fn execute_parameterized_aggregate<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, sql: &'life1 str, params: &'life2 [Value], ) -> Pin<Box<dyn Future<Output = Result<Vec<HashMap<String, Value>>, FraiseQLError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, PostgresAdapter: 'async_trait,

Execute a parameterized aggregate SQL query (GROUP BY / HAVING / window).

fn execute_parameterized_aggregate_with_session<'life0, 'life1, 'life2, 'life3, 'life4, 'life5, 'async_trait>( &'life0 self, sql: &'life1 str, params: &'life2 [Value], session_vars: &'life3 [(&'life4 str, &'life5 str)], routing: ReadRouting, ) -> Pin<Box<dyn Future<Output = Result<Vec<HashMap<String, Value>>, FraiseQLError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, 'life4: 'async_trait, 'life5: 'async_trait, PostgresAdapter: 'async_trait,

Connection-affine variant of execute_parameterized_aggregate.

fn execute_where_query_arc_with_session<'life0, 'life1, 'life2, 'life3, 'life4, 'life5, 'life6, 'async_trait>( &'life0 self, view: &'life1 str, where_clause: Option<&'life2 WhereClause>, limit: Option<u32>, offset: Option<u32>, order_by: Option<&'life3 [OrderByClause]>, session_vars: &'life4 [(&'life5 str, &'life6 str)], routing: ReadRouting, ) -> Pin<Box<dyn Future<Output = Result<Arc<Vec<JsonbValue>>, FraiseQLError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, 'life4: 'async_trait, 'life5: 'async_trait, 'life6: 'async_trait, PostgresAdapter: 'async_trait,

Connection-affine variant of execute_where_query_arc.

fn execute_with_projection_arc_with_session<'life0, 'life1, 'life2, 'life3, 'life4, 'life5, 'async_trait>( &'life0 self, request: &'life1 ProjectionRequest<'life2>, session_vars: &'life3 [(&'life4 str, &'life5 str)], routing: ReadRouting, ) -> Pin<Box<dyn Future<Output = Result<Arc<Vec<JsonbValue>>, FraiseQLError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, 'life4: 'async_trait, 'life5: 'async_trait, PostgresAdapter: 'async_trait,

Connection-affine variant of execute_with_projection_arc.

fn stream_with_projection<'life0, 'life1, 'life2, 'life3, 'life4, 'life5, 'async_trait>( &'life0 self, request: &'life1 ProjectionRequest<'life2>, session_vars: &'life3 [(&'life4 str, &'life5 str)], routing: ReadRouting, ) -> Pin<Box<dyn Future<Output = Result<Pin<Box<dyn Stream<Item = Result<JsonbValue, FraiseQLError>> + Send>>, FraiseQLError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, 'life4: 'async_trait, 'life5: 'async_trait, PostgresAdapter: 'async_trait,

The same read as execute_with_projection_arc_with_session, delivered row by row instead of as a materialised Vec (#958).

fn explain_query<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, sql: &'life1 str, _params: &'life2 [Value], ) -> Pin<Box<dyn Future<Output = Result<Value, FraiseQLError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, PostgresAdapter: 'async_trait,

Run the database’s EXPLAIN on a SQL statement without executing it.

fn query_stats<'life0, 'async_trait>( &'life0 self, limit: u32, ) -> Pin<Box<dyn Future<Output = Result<Vec<QueryStatEntry>, FraiseQLError>> + Send + 'async_trait>>

where 'life0: 'async_trait, PostgresAdapter: 'async_trait,

Retrieve query performance statistics from the database.

fn query_stats_by_id<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<QueryStatEntry>, FraiseQLError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, PostgresAdapter: 'async_trait,

Retrieve statistics for a single query by its ID.

fn reset_query_stats<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<(), FraiseQLError>> + Send + 'async_trait>>

where 'life0: 'async_trait, PostgresAdapter: 'async_trait,

Reset query performance statistics.

fn execute_where_query_arc<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, view: &'life1 str, where_clause: Option<&'life2 WhereClause>, limit: Option<u32>, offset: Option<u32>, order_by: Option<&'life3 [OrderByClause]>, ) -> Pin<Box<dyn Future<Output = Result<Arc<Vec<JsonbValue>>, FraiseQLError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, Self: 'async_trait,

Like execute_where_query but returns the result wrapped in an Arc.

fn execute_with_projection_arc<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, request: &'life1 ProjectionRequest<'life2>, ) -> Pin<Box<dyn Future<Output = Result<Arc<Vec<JsonbValue>>, FraiseQLError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, Self: 'async_trait,

Like execute_with_projection but returns the result wrapped in an Arc.

fn execute_row_query<'life0, 'life1, 'life2, 'life3, 'life4, 'async_trait>( &'life0 self, view_name: &'life1 str, columns: &'life2 [ColumnSpec], where_sql: Option<&'life3 str>, order_by: Option<&'life4 str>, limit: Option<u32>, offset: Option<u32>, ) -> Pin<Box<dyn Future<Output = Result<Vec<Vec<ColumnValue>>, FraiseQLError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, 'life4: 'async_trait, Self: 'async_trait,

Execute a row-shaped query against a view, returning typed column values.

fn invalidate_views<'life0, 'life1, 'async_trait>( &'life0 self, _views: &'life1 [ViewName], ) -> Pin<Box<dyn Future<Output = Result<u64, FraiseQLError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, Self: 'async_trait,

Invalidate cached query results for the specified views.

fn invalidate_by_entity<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, _entity_type: &'life1 str, _entity_id: &'life2 str, ) -> Pin<Box<dyn Future<Output = Result<u64, FraiseQLError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, Self: 'async_trait,

Evict cache entries that contain the given entity UUID.

fn result_cache_stats(&self) -> Option<ResultCacheStats>

A snapshot of this adapter’s query-result cache, or None if it has none.

fn clear_result_cache<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Option<usize>, FraiseQLError>> + Send + 'async_trait>>

where 'life0: 'async_trait, Self: 'async_trait,

Drop every entry in this adapter’s query-result cache.

fn on_schema_reload(&self)

Notify the adapter that the schema has changed.

impl Debug for PostgresAdapter

fn fmt(&self, f: &mut Formatter<'_>) -> Result<(), Error>

Formats the value using the given formatter.

impl RelayDatabaseAdapter for PostgresAdapter

async fn execute_relay_page( &self, view: &str, cursor_column: &str, after: Option<CursorValue>, before: Option<CursorValue>, limit: u32, forward: bool, where_clause: Option<&WhereClause>, order_by: Option<&[OrderByClause]>, include_total_count: bool, ) -> Result<RelayPageResult, FraiseQLError>

Execute keyset (cursor-based) pagination against a JSONB view.

totalCount semantics

When include_total_count is true, two queries are issued on the same connection:

1. A count query — SELECT COUNT(*) FROM {view} WHERE {user_filter} — that reflects the full connection size, ignoring cursor position. This is required by the Relay Cursor Connections spec, which defines totalCount as the count of all objects in the connection, regardless of after/before.

2. A page query — the cursor-filtered, limited result set.

The two-query approach fixes a previous bug where COUNT(*) OVER() ran inside the cursor-filtered subquery, causing totalCount to shrink as the cursor advanced. It also handles the edge case where the current page is empty but the total count is non-zero (e.g., cursor past the last row).

When include_total_count is false, only the page query is issued.

Performance note

The count query scans all rows matching the user filter without LIMIT. On large unfiltered tables this may be slow. Mitigations:

• Only enable totalCount when the client explicitly requests it (enforced by the executor via include_total_count).
• Add a statement_timeout on the connection for relay queries on very large datasets.
• Maintain a denormalised count table or materialised view for hot paths.

async fn execute_relay_page_with_session( &self, view: &str, cursor_column: &str, after: Option<CursorValue>, before: Option<CursorValue>, limit: u32, forward: bool, where_clause: Option<&WhereClause>, order_by: Option<&[OrderByClause]>, include_total_count: bool, session_vars: &[(&str, &str)], routing: ReadRouting, ) -> Result<RelayPageResult, FraiseQLError>

Connection-affine variant of execute_relay_page.

impl Writer for PostgresAdapter

fn execute_write<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, request: &'life1 WriteRequest<'life2>, gate: &'life3 (dyn Fn(&[HashMap<String, Value>]) -> Result<(), FraiseQLError> + Send + Sync), ) -> Pin<Box<dyn Future<Output = Result<Vec<HashMap<String, Value>>, FraiseQLError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, PostgresAdapter: 'async_trait,

Run request under gate (see the trait’s contract).

fn bump_fact_table_versions<'life0, 'life1, 'async_trait>( &'life0 self, _tables: &'life1 [String], ) -> Pin<Box<dyn Future<Output = Result<(), FraiseQLError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, Self: Sync + 'async_trait,

Bump the version counters of tables after a committed write that changed them, so cached aggregates over them are invalidated by version key.