fraiseql_db/traits.rs
1//! Database adapter trait definitions.
2//!
3//! The main [`DatabaseAdapter`] trait lives in this file. Supporting types
4//! (`RelayPageResult`, enums, type aliases) are in
5//! the `adapter_types` submodule.
6
7mod adapter_types;
8mod composed_read;
9mod mutations;
10mod relay;
11
12use std::sync::Arc;
13
14pub use adapter_types::*;
15use async_trait::async_trait;
16pub use composed_read::{
17 COMPOSED_DOCUMENT_KEY, COMPOSED_EMBEDS_KEY, ComposedEmbed, ComposedKeyset, ComposedLevel,
18 EmbedShape, EmbedSource, LevelKeys, composed_read_unsupported,
19};
20use fraiseql_error::{FraiseQLError, Result};
21pub use mutations::{WriteMode, WriteRequest, Writer};
22pub use relay::RelayDatabaseAdapter;
23
24use crate::{
25 types::{
26 DatabaseType, JsonbValue, PoolMetrics, ReadRouting,
27 sql_hints::{OrderByClause, SqlProjectionHint},
28 },
29 where_clause::WhereClause,
30};
31
32/// Adjudicates the rows a mutation function returned, from **inside** the
33/// transaction that produced them and before it commits.
34///
35/// `Ok(())` commits the transaction; `Err(e)` rolls it back and `e` reaches the
36/// caller. See [`Writer::execute_write`] for why a write needs a decision seam this late
37/// (#1353).
38///
39/// Synchronous by construction: the only decision taken here today is the field
40/// authorizer's, whose `authorize_field` is itself synchronous, and holding an open
41/// transaction across an arbitrary `await` would pin a pooled connection for as long
42/// as app-supplied policy code chose to take.
43pub type MutationRowGate<'a> = &'a (
44 dyn Fn(&[std::collections::HashMap<String, serde_json::Value>]) -> Result<()> + Send + Sync
45 );
46
47/// The framework-owned change-log row the mutation executor writes in-txn.
48///
49/// Carries only the fields the adapter cannot derive from the
50/// `app.mutation_response` row it already holds: the DML verb and a NOT-NULL
51/// `object_type` fallback. The changed-entity identity + payload (`object_id`,
52/// `object_data`, `updated_fields`, `cascade`) are read from the function's own
53/// returned row inside the same transaction (see [`Writer::execute_write`]).
54///
55/// This is the Change Spine transactional-outbox contract. Beyond the
56/// `object_type`/`modification_type` + changed-entity columns, it stamps the
57/// envelope: `tenant_id` (carried here, from `SecurityContext`),
58/// `trace_id` (the W3C trace id of the originating request), `schema_version`
59/// (the compiled schema's content hash — a per-deployment constant),
60/// `trace_context` (the full W3C trace context as JSON), `actor_type` /
61/// `acting_for` (the request's actor classification and, for a delegated agent,
62/// the underlying human — #390), `commit_time` (`clock_timestamp()` at INSERT),
63/// and `seq` (the table's `SEQUENCE` default).
64#[derive(Debug, Clone, Copy)]
65pub struct ChangeLogWrite<'a> {
66 /// NOT-NULL fallback for `object_type` when the row's `entity_type` is NULL.
67 /// Sourced from `MutationDefinition.return_type` (always present).
68 pub object_type: &'a str,
69 /// The DML verb written to `modification_type` (e.g. `"INSERT"`,
70 /// `"UPDATE"`, `"DELETE"`, `"CUSTOM"`), from `MutationOperation`.
71 pub modification_type: &'a str,
72 /// The tenant partition stamp written to the `tenant_id UUID` column — the
73 /// Trinity public-facing identifier, read from `SecurityContext.tenant_id`
74 /// at write time and **never** reconstructed from connection / RLS state
75 /// (RLS is PG-only; out-of-session spine consumers bypass it, so the row
76 /// must carry tenant identity explicitly). `None` (→ SQL NULL) for an
77 /// unauthenticated request, a request with no tenant, or a tenant
78 /// identifier that is not a UUID.
79 pub tenant_id: Option<uuid::Uuid>,
80 /// The W3C trace id of the originating request, written to the `trace_id`
81 /// column so an outbox row links back to its distributed trace (the #392
82 /// perf tooling surfaces it as the investigation handle). Read from the
83 /// request's `traceparent` header at write time; `None` (→ SQL NULL) for a
84 /// request without a trace context — e.g. an unauthenticated mutation, which
85 /// carries no `SecurityContext` to stamp.
86 pub trace_id: Option<&'a str>,
87 /// The compiled schema's version written to the `schema_version` column so an
88 /// outbox row records which deployment produced it — the replay /
89 /// zero-downtime correctness handle for #378 (reject a row replayed under a
90 /// different schema). A per-deployment constant derived from the compiled
91 /// schema (`CompiledSchema::content_hash()`), **not** from the request, so it
92 /// changes on any schema change. `None` (→ SQL NULL) for producers with no
93 /// compiled schema in scope — cooperative external producers (ETL) and the
94 /// non-PostgreSQL no-op path.
95 pub schema_version: Option<&'a str>,
96 /// The originating request's **full W3C trace context** as a JSON object
97 /// (`{version, trace_id, parent_id, trace_flags, tracestate?}`), written to the
98 /// `trace_context` JSONB column so a row carries enough to re-propagate /
99 /// reconstruct the distributed trace — not just the scalar `trace_id`. Carried
100 /// here as pre-serialized JSON **text** (the adapter binds it to the JSONB
101 /// column). Built from the request's `traceparent` / `tracestate` headers at
102 /// write time; `None` (→ SQL NULL) for a request without a valid trace context,
103 /// consistent with `trace_id`.
104 pub trace_context: Option<&'a str>,
105 /// The request's actor classification written to the `actor_type` column (the
106 /// `snake_case` `ActorType` token: `"human_user"`, `"service_account"`,
107 /// `"ai_agent"`, `"system_job"`), from `SecurityContext.actor_type()` at write
108 /// time (#390). `None` (→ SQL NULL) for a request with no `SecurityContext` to
109 /// stamp (an unauthenticated mutation), or a cooperative external producer.
110 pub actor_type: Option<&'a str>,
111 /// For a delegated agent request, the **underlying human** the agent acts for
112 /// — the public-facing UUID, written to the `acting_for UUID` column from
113 /// `SecurityContext.acting_for()` (#390). Mirrors `tenant_id`'s UUID shape so
114 /// it is stamped without a DB lookup. `None` (→ SQL NULL) for a non-delegated
115 /// request, an unauthenticated mutation, or a subject that is not UUID-shaped.
116 pub acting_for: Option<uuid::Uuid>,
117 /// The ingress transport the originating request declared (#376), e.g.
118 /// `"mcp"` — merged into the row's `extra_metadata` JSONB as
119 /// `{"transport": …}` so MCP-originated (and, later, other
120 /// transport-tagged) writes are queryable
121 /// (`extra_metadata->>'transport' = 'mcp'`). From
122 /// `SecurityContext.transport()` at write time; `None` omits the key —
123 /// for the HTTP GraphQL path (which does not stamp one today), an
124 /// unauthenticated mutation, or a cooperative producer.
125 pub transport: Option<&'a str>,
126 /// Whether this outbox write also records the changed entity's **pre-image**
127 /// (before-state) into the `object_data_before JSONB` column, sourced from the
128 /// function's own `entity_before` (the after-image comes from `entity`). Set
129 /// from `MutationDefinition.changelog_pre_image`; `false` (the default) leaves
130 /// `object_data_before` out of the INSERT entirely (NULL), byte-for-byte
131 /// today's behavior. The changed-entity payload itself is read from the
132 /// returned row inside the outbox CTE, so this flag only selects which SQL
133 /// form the adapter emits.
134 pub pre_image: bool,
135}
136
137impl<'a> ChangeLogWrite<'a> {
138 /// Build a change-log write descriptor with no envelope stamps (`tenant_id`,
139 /// `trace_id`, `schema_version`, `trace_context`, `actor_type` and
140 /// `acting_for` NULL). Chain [`with_tenant_id`](Self::with_tenant_id) /
141 /// [`with_trace_id`](Self::with_trace_id) /
142 /// [`with_schema_version`](Self::with_schema_version) /
143 /// [`with_trace_context`](Self::with_trace_context) /
144 /// [`with_actor_type`](Self::with_actor_type) /
145 /// [`with_acting_for`](Self::with_acting_for) to stamp them.
146 #[must_use]
147 pub const fn new(object_type: &'a str, modification_type: &'a str) -> Self {
148 Self {
149 object_type,
150 modification_type,
151 tenant_id: None,
152 trace_id: None,
153 schema_version: None,
154 trace_context: None,
155 actor_type: None,
156 acting_for: None,
157 transport: None,
158 pre_image: false,
159 }
160 }
161
162 /// Stamp the tenant partition id (the Trinity public-facing UUID) onto the
163 /// outbox row. `None` leaves `tenant_id` NULL — for system / unauthenticated
164 /// rows, or a tenant identifier that is not UUID-shaped.
165 #[must_use]
166 pub const fn with_tenant_id(mut self, tenant_id: Option<uuid::Uuid>) -> Self {
167 self.tenant_id = tenant_id;
168 self
169 }
170
171 /// Stamp the originating request's W3C trace id onto the outbox row. `None`
172 /// leaves `trace_id` NULL — for a request with no trace context.
173 #[must_use]
174 pub const fn with_trace_id(mut self, trace_id: Option<&'a str>) -> Self {
175 self.trace_id = trace_id;
176 self
177 }
178
179 /// Stamp the compiled schema's version (its content hash) onto the outbox
180 /// row. `None` leaves `schema_version` NULL — for producers with no compiled
181 /// schema in scope (cooperative external producers, the non-PostgreSQL no-op
182 /// path).
183 #[must_use]
184 pub const fn with_schema_version(mut self, schema_version: Option<&'a str>) -> Self {
185 self.schema_version = schema_version;
186 self
187 }
188
189 /// Stamp the originating request's full W3C trace context (pre-serialized JSON
190 /// text) onto the outbox row's `trace_context` JSONB column. `None` leaves it
191 /// NULL — for a request with no valid trace context, or a non-PostgreSQL
192 /// no-op / cooperative producer.
193 #[must_use]
194 pub const fn with_trace_context(mut self, trace_context: Option<&'a str>) -> Self {
195 self.trace_context = trace_context;
196 self
197 }
198
199 /// Stamp the request's actor classification (the `snake_case` `ActorType`
200 /// token) onto the outbox row's `actor_type` column (#390). `None` leaves it
201 /// NULL — for an unauthenticated mutation or a cooperative producer.
202 #[must_use]
203 pub const fn with_actor_type(mut self, actor_type: Option<&'a str>) -> Self {
204 self.actor_type = actor_type;
205 self
206 }
207
208 /// Stamp the delegated user's UUID (the human a delegated agent acts for) onto
209 /// the outbox row's `acting_for` column (#390). `None` leaves it NULL — for a
210 /// non-delegated request, an unauthenticated mutation, or a non-UUID subject.
211 #[must_use]
212 pub const fn with_acting_for(mut self, acting_for: Option<uuid::Uuid>) -> Self {
213 self.acting_for = acting_for;
214 self
215 }
216
217 /// Stamp the ingress transport (e.g. `"mcp"`) onto the outbox row's
218 /// `extra_metadata.transport` key (#376). `None` omits the key — for a
219 /// transport that does not declare itself, an unauthenticated mutation, or
220 /// a cooperative producer.
221 #[must_use]
222 pub const fn with_transport(mut self, transport: Option<&'a str>) -> Self {
223 self.transport = transport;
224 self
225 }
226
227 /// Opt this outbox write into recording the changed entity's pre-image into
228 /// the `object_data_before` column (from the function's `entity_before`). When
229 /// `false` (the default), `object_data_before` is omitted from the INSERT
230 /// entirely, byte-for-byte today's behavior. Set from
231 /// `MutationDefinition.changelog_pre_image`.
232 #[must_use]
233 pub const fn with_pre_image(mut self, pre_image: bool) -> Self {
234 self.pre_image = pre_image;
235 self
236 }
237}
238
239/// Database adapter for executing queries against views.
240///
241/// This trait abstracts over different database backends (PostgreSQL, MySQL, SQLite, SQL Server).
242/// All implementations must support:
243/// - Executing parameterized WHERE queries against views
244/// - Returning JSONB data from the `data` column
245/// - Connection pooling and health checks
246/// - Row-level security (RLS) WHERE clauses
247///
248/// # Architecture
249///
250/// The adapter is the runtime interface to the database. It receives:
251/// - View/table name (e.g., "v_user", "tf_sales")
252/// - Parameterized WHERE clauses (AST form, not strings)
253/// - Projection hints (for performance optimization)
254/// - Pagination parameters (LIMIT/OFFSET)
255///
256/// And returns:
257/// - JSONB rows from the `data` column (most operations)
258/// - Arbitrary rows as HashMap (for aggregation queries)
259/// - Mutation results from stored procedures
260///
261/// # Implementing a New Adapter
262///
263/// To add support for a new database (e.g., Oracle, Snowflake):
264///
265/// 1. **Create a new module** in `src/db/your_database/`
266/// 2. **Implement the trait**:
267///
268/// ```rust,ignore
269/// pub struct YourDatabaseAdapter { /* fields */ }
270///
271/// #[async_trait]
272/// impl DatabaseAdapter for YourDatabaseAdapter {
273/// async fn execute_where_query(&self, ...) -> Result<Vec<JsonbValue>> {
274/// // 1. Build parameterized SQL from WhereClause AST
275/// // 2. Execute with bound parameters (NO string concatenation)
276/// // 3. Return JSONB from data column
277/// }
278/// // Implement other required methods...
279/// }
280/// ```
281/// 3. **Add feature flag** to `Cargo.toml` (e.g., `feature = "your-database"`)
282/// 4. **Copy structure from PostgreSQL adapter** — see `src/db/postgres/adapter.rs`
283/// 5. **Add tests** in `tests/integration/your_database_test.rs`
284///
285/// # Security Requirements
286///
287/// All implementations MUST:
288/// - **Never concatenate user input into SQL strings**
289/// - **Always use parameterized queries** with bind parameters
290/// - **Validate parameter types** before binding
291/// - **Preserve RLS WHERE clauses** (never filter them out)
292/// - **Return errors, not silently fail** (e.g., connection loss)
293///
294/// # Connection Management
295///
296/// - Use a connection pool (recommended: 20 connections default)
297/// - Implement `health_check()` for ping-based monitoring
298/// - Provide `pool_metrics()` for observability
299/// - Handle stale connections gracefully
300///
301/// # Performance Characteristics
302///
303/// Expected throughput when properly implemented:
304/// - **Simple queries** (single table, no WHERE): 250+ Kelem/s
305/// - **Complex queries** (JOINs, multiple conditions): 50+ Kelem/s
306/// - **Mutations** (stored procedures): 1-10 RPS (depends on procedure)
307/// - **Relay pagination** (keyset cursors): 15-30ms latency
308///
309/// # Example: PostgreSQL Implementation
310///
311/// ```rust,ignore
312/// use sqlx::postgres::PgPool;
313/// use async_trait::async_trait;
314///
315/// pub struct PostgresAdapter {
316/// pool: PgPool,
317/// }
318///
319/// #[async_trait]
320/// impl DatabaseAdapter for PostgresAdapter {
321/// async fn execute_where_query(
322/// &self,
323/// view: &str,
324/// where_clause: Option<&WhereClause>,
325/// limit: Option<u32>,
326/// offset: Option<u32>,
327/// ) -> Result<Vec<JsonbValue>> {
328/// // 1. Build SQL: SELECT data FROM {view} WHERE {where_clause} LIMIT {limit}
329/// let mut sql = format!(r#"SELECT data FROM "{}""#, view);
330///
331/// // 2. Add WHERE clause (converts AST to parameterized SQL)
332/// let params = if let Some(where_clause) = where_clause {
333/// sql.push_str(" WHERE ");
334/// let (where_sql, params) = build_where_sql(where_clause)?;
335/// sql.push_str(&where_sql);
336/// params
337/// } else {
338/// vec![]
339/// };
340///
341/// // 3. Add LIMIT and OFFSET
342/// if let Some(limit) = limit {
343/// sql.push_str(" LIMIT ");
344/// sql.push_str(&limit.to_string());
345/// }
346/// if let Some(offset) = offset {
347/// sql.push_str(" OFFSET ");
348/// sql.push_str(&offset.to_string());
349/// }
350///
351/// // 4. Execute with bound parameters (NO string interpolation)
352/// let rows: Vec<(serde_json::Value,)> = sqlx::query_as(&sql)
353/// .bind(¶ms[0])
354/// .bind(¶ms[1])
355/// // ... bind all parameters
356/// .fetch_all(&self.pool)
357/// .await?;
358///
359/// // 5. Extract JSONB and return
360/// Ok(rows.into_iter().map(|(data,)| data).collect())
361/// }
362///
363/// // Implement other required methods...
364/// }
365/// ```
366///
367/// # Example: Basic Usage
368///
369/// ```rust,no_run
370/// use fraiseql_db::{DatabaseAdapter, WhereClause, WhereOperator};
371/// use serde_json::json;
372///
373/// # async fn example(adapter: impl DatabaseAdapter) -> Result<(), Box<dyn std::error::Error>> {
374/// // Build WHERE clause (AST, not string)
375/// let where_clause = WhereClause::Field {
376/// path: vec!["email".to_string()],
377/// operator: WhereOperator::Icontains,
378/// value: json!("example.com"),
379/// };
380///
381/// // Execute query with parameters
382/// let results = adapter
383/// .execute_where_query("v_user", Some(&where_clause), Some(10), None, None)
384/// .await?;
385///
386/// println!("Found {} users matching filter", results.len());
387/// # Ok(())
388/// # }
389/// ```
390///
391/// # See Also
392///
393/// - `WhereClause` — AST for parameterized WHERE clauses
394/// - `RelayDatabaseAdapter` — Optional trait for keyset pagination
395/// - [Performance Guide](https://docs.fraiseql.rs/performance/database-adapters.md)
396// POLICY: `#[async_trait]` placement for `DatabaseAdapter`
397//
398// `DatabaseAdapter` is used both generically (`Server<A: DatabaseAdapter>` in axum
399// handlers, zero overhead via static dispatch) and dynamically (`Arc<dyn
400// DatabaseAdapter + Send + Sync>` in federation, heap-boxed future per call).
401//
402// `#[async_trait]` is required on:
403// - The trait definition (generates `Pin<Box<dyn Future + Send>>` return types)
404// - Every `impl DatabaseAdapter for ConcreteType` block (generates the boxing)
405// NOT required on callers (they see `Pin<Box<dyn Future + Send>>` from macro output).
406//
407// Why not native `async fn in trait` (Rust 1.75+)?
408// Native dyn async trait does NOT propagate `+ Send` on generated futures. Tokio
409// requires futures spawned with `tokio::spawn` to be `Send`. Until Return Type
410// Notation (RFC 3425, tracking: github.com/rust-lang/rust/issues/109417) stabilises,
411// `async_trait` is the only ergonomic path to `dyn DatabaseAdapter + Send + Sync`.
412// Re-evaluate when Rust 1.90+ ships or when RTN is stabilised.
413//
414// MIGRATION TRACKING: async-trait → native async fn in trait
415//
416// Current status: BLOCKED on RFC 3425 (Return Type Notation)
417// See: https://github.com/rust-lang/rfcs/pull/3425
418// https://github.com/rust-lang/rust/issues/109417
419//
420// Migration is safe when ALL of the following are true:
421// 1. RTN with `+ Send` bounds is stable on rustc (e.g. `fn foo() -> impl Future + Send`)
422// 2. FraiseQL MSRV is updated to that stabilising version
423// 3. tokio::spawn() works with native dyn async trait objects (futures must be Send)
424//
425// Scope when criteria are met: 68 files (grep -rn "#\[async_trait\]" crates/)
426// Effort: Medium (mostly mechanical — remove macro from impls, adjust trait defs)
427// dynosaur was evaluated and rejected: does not propagate + Send (incompatible with Tokio)
428#[async_trait]
429pub trait DatabaseAdapter: Send + Sync + 'static {
430 /// Execute a WHERE query against a view and return JSONB rows.
431 ///
432 /// # Arguments
433 ///
434 /// * `view` - View name (e.g., "v_user", "v_post")
435 /// * `where_clause` - Optional WHERE clause AST
436 /// * `limit` - Optional row limit (for pagination)
437 /// * `offset` - Optional row offset (for pagination)
438 /// * `security_context` - Optional security context for RLS and caching decisions
439 ///
440 /// # Returns
441 ///
442 /// Vec of JSONB values from the `data` column.
443 ///
444 /// # Errors
445 ///
446 /// Returns `FraiseQLError::Database` on query execution failure.
447 /// Returns `FraiseQLError::ConnectionPool` if connection pool is exhausted.
448 ///
449 /// # Example
450 ///
451 /// ```rust,no_run
452 /// # use fraiseql_db::DatabaseAdapter;
453 /// # async fn example(adapter: impl DatabaseAdapter) -> Result<(), Box<dyn std::error::Error>> {
454 /// // Simple query without WHERE clause
455 /// let all_users = adapter
456 /// .execute_where_query("v_user", None, Some(10), Some(0), None)
457 /// .await?;
458 /// # Ok(())
459 /// # }
460 /// ```
461 async fn execute_where_query(
462 &self,
463 view: &str,
464 where_clause: Option<&WhereClause>,
465 limit: Option<u32>,
466 offset: Option<u32>,
467 order_by: Option<&[OrderByClause]>,
468 ) -> Result<Vec<JsonbValue>>;
469
470 /// Execute a WHERE query with SQL field projection optimization.
471 ///
472 /// Projects only the requested fields at the database level, reducing network payload
473 /// and JSON deserialization overhead by **40-55%** based on production measurements.
474 ///
475 /// This is the primary query execution method for optimized GraphQL queries.
476 /// It automatically selects only the fields requested in the GraphQL query, avoiding
477 /// unnecessary network transfer and deserialization of unused fields.
478 ///
479 /// # Automatic Projection
480 ///
481 /// In most cases, you don't call this directly. The `Executor` automatically:
482 /// 1. Determines which fields the GraphQL query requests
483 /// 2. Generates a `SqlProjectionHint` using database-specific SQL
484 /// 3. Calls this method with the projection hint
485 ///
486 /// # Arguments
487 ///
488 /// * `view` - View name (e.g., "v_user", "v_post")
489 /// * `projection` - Optional SQL projection hint with field list
490 /// - `Some(hint)`: Use projection to select only requested fields
491 /// - `None`: Falls back to standard query (full JSONB column)
492 /// * `where_clause` - Optional WHERE clause AST for filtering
493 /// * `limit` - Optional row limit (for pagination)
494 ///
495 /// # Returns
496 ///
497 /// Vec of JSONB values, either:
498 /// - Full objects (when projection is None)
499 /// - Projected objects with only requested fields (when projection is Some)
500 ///
501 /// # Errors
502 ///
503 /// Returns `FraiseQLError::Database` on query execution failure, including:
504 /// - Connection pool exhaustion
505 /// - SQL execution errors
506 /// - Type mismatches
507 ///
508 /// # Performance Characteristics
509 ///
510 /// When projection is provided (recommended):
511 /// - **Latency**: 40-55% reduction vs full object fetch
512 /// - **Network**: 40-55% smaller payload (proportional to unused fields)
513 /// - **Throughput**: Maintains 250+ Kelem/s (elements per second)
514 /// - **Memory**: Proportional to projected fields only
515 ///
516 /// Improvement scales with:
517 /// - Percentage of unused fields (more unused = more improvement)
518 /// - Size of result set (larger sets benefit more)
519 /// - Network latency (network-bound queries benefit most)
520 ///
521 /// When projection is None:
522 /// - Behavior identical to `execute_where_query()`
523 /// - Returns full JSONB column
524 /// - Used for compatibility/debugging
525 ///
526 /// # Database Support
527 ///
528 /// | Database | Status | Implementation |
529 /// |----------|--------|-----------------|
530 /// | PostgreSQL | ✅ Optimized | `jsonb_build_object()` |
531 /// | MySQL | ⏳ Fallback | Server-side filtering (planned) |
532 /// | SQLite | ⏳ Fallback | Server-side filtering (planned) |
533 /// | SQL Server | ⏳ Fallback | Server-side filtering (planned) |
534 ///
535 /// # Example: Direct Usage (Advanced)
536 ///
537 /// ```no_run
538 /// // Requires: running PostgreSQL database and a DatabaseAdapter implementation.
539 /// use fraiseql_db::types::SqlProjectionHint;
540 /// use fraiseql_db::traits::DatabaseAdapter;
541 /// use fraiseql_db::DatabaseType;
542 ///
543 /// # async fn example(adapter: &impl DatabaseAdapter) -> Result<(), Box<dyn std::error::Error>> {
544 /// let projection = SqlProjectionHint::new(
545 /// DatabaseType::PostgreSQL,
546 /// "jsonb_build_object(\
547 /// 'id', data->>'id', \
548 /// 'name', data->>'name', \
549 /// 'email', data->>'email'\
550 /// )".to_string(),
551 /// 75,
552 /// );
553 ///
554 /// let results = adapter
555 /// .execute_with_projection("v_user", Some(&projection), None, Some(100), None, None)
556 /// .await?;
557 ///
558 /// // results only contain id, name, email fields
559 /// // 75% smaller than fetching all fields
560 /// # Ok(())
561 /// # }
562 /// ```
563 ///
564 /// # Example: Fallback (No Projection)
565 ///
566 /// ```no_run
567 /// // Requires: running PostgreSQL database and a DatabaseAdapter implementation.
568 /// # use fraiseql_db::traits::DatabaseAdapter;
569 /// # async fn example(adapter: &impl DatabaseAdapter) -> Result<(), Box<dyn std::error::Error>> {
570 /// // For debugging or when projection not available
571 /// let results = adapter
572 /// .execute_with_projection("v_user", None, None, Some(100), None, None)
573 /// .await?;
574 ///
575 /// // Equivalent to execute_where_query() - returns full objects
576 /// # Ok(())
577 /// # }
578 /// ```
579 ///
580 /// # See Also
581 ///
582 /// - `execute_where_query()` - Standard query without projection
583 /// - `SqlProjectionHint` - Structure defining field projection
584 /// - [Projection Optimization Guide](https://docs.fraiseql.rs/performance/projection-optimization.md)
585 async fn execute_with_projection(
586 &self,
587 view: &str,
588 projection: Option<&SqlProjectionHint>,
589 where_clause: Option<&WhereClause>,
590 limit: Option<u32>,
591 offset: Option<u32>,
592 order_by: Option<&[OrderByClause]>,
593 ) -> Result<Vec<JsonbValue>>;
594
595 /// Like `execute_where_query` but returns the result wrapped in an `Arc`.
596 ///
597 /// The default implementation wraps the result of `execute_where_query` in a
598 /// fresh `Arc`. `CachedDatabaseAdapter` overrides this to return the cached `Arc`
599 /// directly — eliminating the full `Vec<JsonbValue>` clone that the non-`Arc`
600 /// path requires on every cache hit.
601 ///
602 /// Callers on the hot query path should prefer this variant and borrow from the
603 /// `Arc` via `&**arc` rather than taking ownership.
604 ///
605 /// # Errors
606 ///
607 /// Same errors as `execute_where_query`.
608 async fn execute_where_query_arc(
609 &self,
610 view: &str,
611 where_clause: Option<&WhereClause>,
612 limit: Option<u32>,
613 offset: Option<u32>,
614 order_by: Option<&[OrderByClause]>,
615 ) -> Result<Arc<Vec<JsonbValue>>> {
616 self.execute_where_query(view, where_clause, limit, offset, order_by)
617 .await
618 .map(Arc::new)
619 }
620
621 /// Like `execute_with_projection` but returns the result wrapped in an `Arc`.
622 ///
623 /// The default implementation wraps the result of `execute_with_projection` in a
624 /// fresh `Arc`. `CachedDatabaseAdapter` overrides this to return the cached `Arc`
625 /// directly — eliminating the full `Vec<JsonbValue>` clone that the non-`Arc`
626 /// path requires on every cache hit.
627 ///
628 /// Parameters are passed in a `ProjectionRequest` struct (F043) so adapters
629 /// and callers cannot misorder them.
630 ///
631 /// # Errors
632 ///
633 /// Same errors as `execute_with_projection`.
634 async fn execute_with_projection_arc(
635 &self,
636 request: &ProjectionRequest<'_>,
637 ) -> Result<Arc<Vec<JsonbValue>>> {
638 self.execute_with_projection(
639 request.view,
640 request.projection,
641 request.where_clause,
642 request.limit,
643 request.offset,
644 request.order_by,
645 )
646 .await
647 .map(Arc::new)
648 }
649
650 /// Get database type (for logging/metrics).
651 ///
652 /// Used to identify which database backend is in use.
653 fn database_type(&self) -> DatabaseType;
654
655 /// Health check - verify database connectivity.
656 ///
657 /// Executes a simple query (e.g., `SELECT 1`) to verify the database is reachable.
658 ///
659 /// # Errors
660 ///
661 /// Returns `FraiseQLError::Database` if health check fails.
662 async fn health_check(&self) -> Result<()>;
663
664 /// Get connection pool metrics.
665 ///
666 /// Returns current statistics about the connection pool:
667 /// - Total connections
668 /// - Idle connections
669 /// - Active connections
670 /// - Waiting requests
671 fn pool_metrics(&self) -> PoolMetrics;
672
673 /// Execute raw SQL query and return rows as JSON objects.
674 ///
675 /// Used for aggregation queries where we need full row data, not just JSONB column.
676 ///
677 /// # Security Warning
678 ///
679 /// This method executes arbitrary SQL. **NEVER** pass untrusted input directly to this method.
680 /// Always:
681 /// - Use parameterized queries with bound parameters
682 /// - Validate and sanitize SQL templates before execution
683 /// - Only execute SQL generated by the FraiseQL compiler
684 /// - Log SQL execution for audit trails
685 ///
686 /// # Arguments
687 ///
688 /// * `sql` - Raw SQL query to execute (must be safe/trusted)
689 ///
690 /// # Returns
691 ///
692 /// Vec of rows, where each row is a HashMap of column name to JSON value.
693 ///
694 /// # Errors
695 ///
696 /// Returns `FraiseQLError::Database` on query execution failure.
697 ///
698 /// # Example
699 ///
700 /// ```rust,no_run
701 /// # use fraiseql_db::DatabaseAdapter;
702 /// # async fn example(adapter: impl DatabaseAdapter) -> Result<(), Box<dyn std::error::Error>> {
703 /// // Safe: SQL generated by FraiseQL compiler
704 /// let sql = "SELECT category, SUM(revenue) as total FROM tf_sales GROUP BY category";
705 /// let rows = adapter.execute_raw_query(sql).await?;
706 /// for row in rows {
707 /// println!("Category: {}, Total: {}", row["category"], row["total"]);
708 /// }
709 /// # Ok(())
710 /// # }
711 /// ```
712 async fn execute_raw_query(
713 &self,
714 sql: &str,
715 ) -> Result<Vec<std::collections::HashMap<String, serde_json::Value>>>;
716
717 /// Execute a row-shaped query against a view, returning typed column values.
718 ///
719 /// Used by the gRPC transport for protobuf encoding of query results.
720 /// The default implementation delegates to `execute_raw_query` and converts
721 /// JSON results to `ColumnValue` vectors.
722 ///
723 /// # Errors
724 ///
725 /// Returns `FraiseQLError::Database` if the adapter returns an error.
726 async fn execute_row_query(
727 &self,
728 view_name: &str,
729 columns: &[crate::types::ColumnSpec],
730 where_sql: Option<&str>,
731 order_by: Option<&str>,
732 limit: Option<u32>,
733 offset: Option<u32>,
734 ) -> Result<Vec<Vec<crate::types::ColumnValue>>> {
735 let sql = build_row_query_sql(view_name, where_sql, order_by, limit, offset);
736 let results = self.execute_raw_query(&sql).await?;
737
738 Ok(results.iter().map(|row| row_to_column_values(row, columns)).collect())
739 }
740
741 /// Connection-affine variant of [`execute_row_query`](Self::execute_row_query).
742 ///
743 /// See
744 /// [`execute_where_query_arc_with_session`](Self::execute_where_query_arc_with_session)
745 /// for the rationale: an RLS policy backed by `current_setting()` (#329) only sees
746 /// the variables if they are applied transaction-locally on the **same** connection
747 /// as the read.
748 ///
749 /// # Why the row shape needs its own pair
750 ///
751 /// The row-shaped read is the gRPC transport's source. Until #1351 it resolved its
752 /// own predicate and never reached the engine, so it had no session variables to
753 /// pin and none of these methods existed. Routing it through the direct-read
754 /// chokepoint gives it `session_variables` like every other read — and a resolved
755 /// value the read cannot apply is the failure mode the chokepoint exists to
756 /// prevent, so it travels rather than being dropped.
757 ///
758 /// # Errors
759 ///
760 /// Same errors as [`execute_row_query`](Self::execute_row_query); additionally
761 /// returns `FraiseQLError::Database` if `set_config` fails on any pair.
762 async fn execute_row_query_with_session(
763 &self,
764 view_name: &str,
765 columns: &[crate::types::ColumnSpec],
766 where_sql: Option<&str>,
767 order_by: Option<&str>,
768 limit: Option<u32>,
769 offset: Option<u32>,
770 _session_vars: &[(&str, &str)],
771 ) -> Result<Vec<Vec<crate::types::ColumnValue>>> {
772 self.execute_row_query(view_name, columns, where_sql, order_by, limit, offset)
773 .await
774 }
775
776 /// Execute a parameterized aggregate SQL query (GROUP BY / HAVING / window).
777 ///
778 /// `sql` contains `$N` (PostgreSQL), `?` (MySQL / SQLite), or `@P1` (SQL Server)
779 /// placeholders for string and array values; numeric and NULL values may be inlined.
780 /// `params` are the corresponding values in placeholder order.
781 ///
782 /// Unlike `execute_raw_query`, this method accepts bind parameters so that
783 /// user-supplied filter values never appear as string literals in the SQL text,
784 /// eliminating the injection risk that `escape_sql_string` mitigated previously.
785 ///
786 /// # Arguments
787 ///
788 /// * `sql` - SQL with placeholders generated by
789 /// `AggregationSqlGenerator::generate_parameterized`
790 /// * `params` - Bind parameters in placeholder order
791 ///
792 /// # Returns
793 ///
794 /// Vec of rows, where each row is a `HashMap` of column name to JSON value.
795 ///
796 /// # Errors
797 ///
798 /// Returns `FraiseQLError::Database` on execution failure.
799 /// Returns `FraiseQLError::Database` on adapters that do not support raw SQL
800 /// (e.g., `FraiseWireAdapter`).
801 async fn execute_parameterized_aggregate(
802 &self,
803 sql: &str,
804 params: &[serde_json::Value],
805 ) -> Result<Vec<std::collections::HashMap<String, serde_json::Value>>>;
806
807 /// Connection-affine variant of
808 /// [`execute_parameterized_aggregate`](Self::execute_parameterized_aggregate).
809 ///
810 /// Applies `session_vars` transaction-locally on the same connection that
811 /// runs the aggregate, so aggregate views backed by `current_setting()` RLS
812 /// observe the configured values (fixes #329 for the aggregate path). The default
813 /// ignores `session_vars`: a backend without transaction-local settings has none to
814 /// apply.
815 ///
816 /// # Errors
817 ///
818 /// Same errors as [`execute_parameterized_aggregate`](Self::execute_parameterized_aggregate);
819 /// additionally returns `FraiseQLError::Database` if `set_config` fails.
820 async fn execute_parameterized_aggregate_with_session(
821 &self,
822 sql: &str,
823 params: &[serde_json::Value],
824 _session_vars: &[(&str, &str)],
825 _routing: ReadRouting,
826 ) -> Result<Vec<std::collections::HashMap<String, serde_json::Value>>> {
827 self.execute_parameterized_aggregate(sql, params).await
828 }
829
830 /// Whether this adapter implements
831 /// [`execute_composed_with_session`](Self::execute_composed_with_session) — whether a
832 /// read can have related resources composed into it (the REST `?select=` embed).
833 ///
834 /// Asked twice, of one answer: when the REST transport is mounted, so a deployment
835 /// whose adapter cannot serve embeds is told at boot rather than by its first embed;
836 /// and by the engine before a composed read is sent, so the `501` a request gets is
837 /// that same fact rather than a second one.
838 ///
839 /// **Implementing `execute_composed_with_session` obliges you to override this too**,
840 /// and a wrapping adapter must forward both. Defaults to `false`: a capability is not
841 /// something a backend should acquire by omission.
842 fn supports_composed_reads(&self) -> bool {
843 false
844 }
845
846 /// Invalidate cached query results for the specified views.
847 ///
848 /// Called by the executor after a mutation succeeds, so that stale cache
849 /// entries reading from modified views are evicted. The default
850 /// implementation is a no-op; `CachedDatabaseAdapter` overrides this.
851 ///
852 /// View names are passed as `&[ViewName]` so the wrapper's `Arc<str>`
853 /// backing is preserved across the call. Callers that hold a `String`
854 /// can convert in place with `ViewName::from(...)`.
855 ///
856 /// # Returns
857 ///
858 /// The number of cache entries evicted.
859 async fn invalidate_views(&self, _views: &[crate::ViewName]) -> Result<u64> {
860 Ok(0)
861 }
862
863 /// Evict cache entries that contain the given entity UUID.
864 ///
865 /// Called by the executor after a successful UPDATE or DELETE mutation when
866 /// the `mutation_response` includes an `entity_id`. Only cache entries whose
867 /// entity-ID index contains the given UUID are removed; unrelated entries
868 /// remain warm.
869 ///
870 /// The default implementation is a no-op. `CachedDatabaseAdapter` overrides
871 /// this to perform the selective eviction.
872 ///
873 /// # Returns
874 ///
875 /// The number of cache entries evicted.
876 async fn invalidate_by_entity(&self, _entity_type: &str, _entity_id: &str) -> Result<u64> {
877 Ok(0)
878 }
879
880 /// A snapshot of this adapter's query-result cache, or `None` if it has none.
881 ///
882 /// The operator surface for the cache that actually serves GraphQL queries.
883 /// `None` means "this adapter does not cache" — distinct from `Some(stats)` with
884 /// zero entries, which means "the cache is there and currently empty". The admin
885 /// API reported the second as the first for every non-Arrow deployment (#941).
886 ///
887 /// The default implementation returns `None`; `CachedDatabaseAdapter` overrides it.
888 fn result_cache_stats(&self) -> Option<crate::ResultCacheStats> {
889 None
890 }
891
892 /// Drop every entry in this adapter's query-result cache.
893 ///
894 /// Returns the number of entries dropped, or `None` if this adapter has no cache
895 /// — again so the caller can say "there is no such cache" rather than reporting a
896 /// successful clear of nothing.
897 ///
898 /// The default implementation returns `Ok(None)`; `CachedDatabaseAdapter` overrides it.
899 ///
900 /// # Errors
901 ///
902 /// Returns `FraiseQLError` if the underlying cache cannot be cleared.
903 async fn clear_result_cache(&self) -> Result<Option<usize>> {
904 Ok(None)
905 }
906
907 /// Run the database's `EXPLAIN` on a SQL statement without executing it.
908 ///
909 /// Returns a JSON representation of the query plan. The format is
910 /// database-specific (e.g. PostgreSQL returns JSON, SQLite returns rows).
911 ///
912 /// The default implementation returns `Unsupported`.
913 async fn explain_query(
914 &self,
915 _sql: &str,
916 _params: &[serde_json::Value],
917 ) -> Result<serde_json::Value> {
918 Err(fraiseql_error::FraiseQLError::Unsupported {
919 message: "EXPLAIN not available for this database adapter".to_string(),
920 })
921 }
922
923 /// Run `EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON)` against a view with the
924 /// same parameterized WHERE clause that `execute_where_query` would use.
925 ///
926 /// Unlike `explain_query`, this method uses **real bound parameters** and
927 /// **actually executes the query** (ANALYZE mode), so the plan reflects
928 /// PostgreSQL's runtime statistics for the given filter values.
929 ///
930 /// Only PostgreSQL supports this; other adapters return
931 /// `FraiseQLError::Unsupported` by default.
932 ///
933 /// # Arguments
934 ///
935 /// * `view` - View name (e.g., "v_user")
936 /// * `where_clause` - Optional filter (same as `execute_where_query`)
937 /// * `limit` - Optional row limit
938 /// * `offset` - Optional row offset
939 ///
940 /// # Errors
941 ///
942 /// Returns `FraiseQLError::Database` on execution failure.
943 /// Returns `FraiseQLError::Unsupported` for non-PostgreSQL adapters.
944 async fn explain_where_query(
945 &self,
946 _view: &str,
947 _where_clause: Option<&WhereClause>,
948 _limit: Option<u32>,
949 _offset: Option<u32>,
950 ) -> Result<serde_json::Value> {
951 Err(fraiseql_error::FraiseQLError::Unsupported {
952 message: "EXPLAIN ANALYZE is not available for this database adapter. \
953 Only PostgreSQL supports explain_where_query."
954 .to_string(),
955 })
956 }
957
958 /// Connection-affine variant of [`execute_where_query_arc`](Self::execute_where_query_arc).
959 ///
960 /// Applies `session_vars` transaction-locally on the same connection that
961 /// runs the read, so PostgreSQL Row-Level-Security policies backed by
962 /// `current_setting()` see the configured values (fixes #329). The default ignores
963 /// `session_vars`: a backend without transaction-local settings has none to apply.
964 ///
965 /// # Errors
966 ///
967 /// Same errors as [`execute_where_query_arc`](Self::execute_where_query_arc); additionally
968 /// returns `FraiseQLError::Database` if `set_config` fails on any pair.
969 async fn execute_where_query_arc_with_session(
970 &self,
971 view: &str,
972 where_clause: Option<&WhereClause>,
973 limit: Option<u32>,
974 offset: Option<u32>,
975 order_by: Option<&[OrderByClause]>,
976 _session_vars: &[(&str, &str)],
977 _routing: ReadRouting,
978 ) -> Result<Arc<Vec<JsonbValue>>> {
979 self.execute_where_query_arc(view, where_clause, limit, offset, order_by).await
980 }
981
982 /// Count the rows of `view` matching `where_clause`, without materialising them.
983 ///
984 /// Backs both the GraphQL `<name>Count` sibling (#938) and the REST
985 /// `Prefer: count=exact` header, so the two cannot answer the same question
986 /// differently.
987 ///
988 /// `session_vars` are applied transaction-locally on the connection that
989 /// runs the count, exactly as for the read it describes (#329). This is
990 /// load-bearing rather than symmetric-for-neatness: under RLS, a count that
991 /// ran without the session variables would report the *unfiltered* total
992 /// beside a filtered page — a row-count oracle over rows the caller cannot
993 /// read.
994 ///
995 /// # Default implementation
996 ///
997 /// The default fetches the matching rows and counts them. That is *correct*
998 /// — same view, same predicate, same session — but it is O(rows) in memory,
999 /// so an adapter that can push the count into the database should override
1000 /// it. `PostgresAdapter` does, with `SELECT COUNT(*)`.
1001 ///
1002 /// # Errors
1003 ///
1004 /// Returns `FraiseQLError::Database` if the query fails.
1005 async fn count_where_query(
1006 &self,
1007 view: &str,
1008 where_clause: Option<&WhereClause>,
1009 session_vars: &[(&str, &str)],
1010 routing: ReadRouting,
1011 ) -> Result<u64> {
1012 let rows = self
1013 .execute_where_query_arc_with_session(
1014 view,
1015 where_clause,
1016 None,
1017 None,
1018 None,
1019 session_vars,
1020 routing,
1021 )
1022 .await?;
1023 Ok(rows.len() as u64)
1024 }
1025
1026 /// Connection-affine variant of
1027 /// [`execute_with_projection_arc`](Self::execute_with_projection_arc).
1028 ///
1029 /// See [`execute_where_query_arc_with_session`](Self::execute_where_query_arc_with_session) for
1030 /// the rationale.
1031 ///
1032 /// # Errors
1033 ///
1034 /// Same errors as [`execute_with_projection_arc`](Self::execute_with_projection_arc);
1035 /// additionally returns `FraiseQLError::Database` if `set_config` fails on any pair.
1036 async fn execute_with_projection_arc_with_session(
1037 &self,
1038 request: &ProjectionRequest<'_>,
1039 _session_vars: &[(&str, &str)],
1040 _routing: ReadRouting,
1041 ) -> Result<Arc<Vec<JsonbValue>>> {
1042 self.execute_with_projection_arc(request).await
1043 }
1044
1045 /// The same read as
1046 /// [`execute_with_projection_arc_with_session`](Self::execute_with_projection_arc_with_session),
1047 /// delivered row by row instead of as a materialised `Vec` (#958).
1048 ///
1049 /// Same view, same predicate, same session variables, same routing. The only
1050 /// difference is *when* the rows arrive and how much memory the caller needs
1051 /// to hold them: a streamed read is `O(1)` in the number of rows, so an export
1052 /// of a table larger than memory is possible where the collecting method would
1053 /// have to fail.
1054 ///
1055 /// # Why this exists as its own method
1056 ///
1057 /// A paginated re-execution loop (`LIMIT n OFFSET k`, `k += n`) delivers the
1058 /// same rows in bounded memory and needs no new trait method, and that is what
1059 /// the export surfaces did. It has two properties a stream does not:
1060 ///
1061 /// - it is `O(offset)` per batch, so exporting `N` rows costs `O(N²)` row scans;
1062 /// - each batch is its own snapshot, so a concurrent insert or delete between batches shifts
1063 /// rows across the page boundary — silently duplicating a row into two batches, or skipping
1064 /// one entirely.
1065 ///
1066 /// One statement over one portal has neither problem.
1067 ///
1068 /// # Default implementation
1069 ///
1070 /// Collects the read and replays it. That is *correct* — same rows, same order —
1071 /// but it holds the whole result set, so an adapter that can stream should
1072 /// override it. `PostgresAdapter` does.
1073 ///
1074 /// ⚠ A wrapping adapter (caching, instrumentation) that inherits this default
1075 /// converts every streaming caller back into a buffering one **without failing**:
1076 /// the rows are right and the memory bound is gone. A wrapper must forward this
1077 /// method explicitly.
1078 ///
1079 /// # Errors
1080 ///
1081 /// Returns the same errors as
1082 /// [`execute_with_projection_arc_with_session`](Self::execute_with_projection_arc_with_session).
1083 /// Note that an implementation that truly streams can only report the errors it
1084 /// has seen *so far* when this returns: a failure raised after the first row —
1085 /// a lost connection, a runtime cast error on a later row — surfaces as an `Err`
1086 /// item inside the stream, not from this call.
1087 async fn stream_with_projection(
1088 &self,
1089 request: &ProjectionRequest<'_>,
1090 session_vars: &[(&str, &str)],
1091 routing: ReadRouting,
1092 ) -> Result<JsonbRowStream> {
1093 let rows = self
1094 .execute_with_projection_arc_with_session(request, session_vars, routing)
1095 .await?;
1096 // Unwrap when this call owns the only reference (the usual case, since the
1097 // read above just built it) so replaying the buffer costs no deep clone.
1098 let rows = Arc::try_unwrap(rows).unwrap_or_else(|shared| (*shared).clone());
1099 Ok(Box::pin(futures::stream::iter(rows.into_iter().map(Ok))))
1100 }
1101
1102 /// Execute a read that composes its embedded levels into **one** statement.
1103 ///
1104 /// The REST `?select=` embed used to be a parent read plus one sub-read per parent
1105 /// row per level; this is the same answer in one round trip, one snapshot and one
1106 /// pooled connection. See [`ComposedLevel`] for the call shape and for the
1107 /// `{"d": …, "e": …}` rows it returns.
1108 ///
1109 /// Session variables and routing mean what they mean for
1110 /// [`execute_with_projection_arc_with_session`](Self::execute_with_projection_arc_with_session).
1111 ///
1112 /// # Default implementation
1113 ///
1114 /// Refuses. Composing needs correlated `LATERAL` subqueries and JSON aggregation,
1115 /// which an adapter has to render for its own dialect; answering with anything
1116 /// else — the root rows without their embeds, say — would be a partial response
1117 /// indistinguishable from a parent with no related rows. A wrapping adapter must
1118 /// forward this method explicitly.
1119 ///
1120 /// # Errors
1121 ///
1122 /// `FraiseQLError::Unsupported` from the default; otherwise the errors of
1123 /// [`execute_with_projection_arc_with_session`](Self::execute_with_projection_arc_with_session).
1124 async fn execute_composed_with_session(
1125 &self,
1126 read: &ComposedLevel,
1127 _session_vars: &[(&str, &str)],
1128 _routing: ReadRouting,
1129 ) -> Result<Arc<Vec<JsonbValue>>> {
1130 Err(composed_read_unsupported(&read.view))
1131 }
1132
1133 /// The same read as [`execute_row_query`](Self::execute_row_query), delivered
1134 /// row by row (#958).
1135 ///
1136 /// The column-shaped counterpart of
1137 /// [`stream_with_projection`](Self::stream_with_projection); see that method for
1138 /// why streaming and paginated re-execution are not interchangeable. This is the
1139 /// gRPC transport's source, which encodes one protobuf message per row and has
1140 /// no use for a `Vec` of them.
1141 ///
1142 /// # Default implementation
1143 ///
1144 /// Collects and replays, with the same caveat about wrapping adapters.
1145 ///
1146 /// # Errors
1147 ///
1148 /// Returns the same errors as [`execute_row_query`](Self::execute_row_query);
1149 /// see [`stream_with_projection`](Self::stream_with_projection) for which of
1150 /// them can still arrive after this call has returned `Ok`.
1151 async fn stream_row_query(
1152 &self,
1153 view_name: &str,
1154 columns: &[crate::types::ColumnSpec],
1155 where_sql: Option<&str>,
1156 order_by: Option<&str>,
1157 limit: Option<u32>,
1158 offset: Option<u32>,
1159 ) -> Result<ColumnRowStream> {
1160 let rows = self
1161 .execute_row_query(view_name, columns, where_sql, order_by, limit, offset)
1162 .await?;
1163 Ok(Box::pin(futures::stream::iter(rows.into_iter().map(Ok))))
1164 }
1165
1166 /// Connection-affine variant of [`stream_row_query`](Self::stream_row_query).
1167 ///
1168 /// The streaming twin of
1169 /// [`execute_row_query_with_session`](Self::execute_row_query_with_session); see
1170 /// that method for why the row shape carries session variables at all. The
1171 /// streaming arm is where dropping them costs the most, because it is the arm
1172 /// that delivers an unbounded number of rows.
1173 ///
1174 /// # Errors
1175 ///
1176 /// Same errors as [`stream_row_query`](Self::stream_row_query); additionally
1177 /// returns `FraiseQLError::Database` if `set_config` fails on any pair.
1178 async fn stream_row_query_with_session(
1179 &self,
1180 view_name: &str,
1181 columns: &[crate::types::ColumnSpec],
1182 where_sql: Option<&str>,
1183 order_by: Option<&str>,
1184 limit: Option<u32>,
1185 offset: Option<u32>,
1186 _session_vars: &[(&str, &str)],
1187 ) -> Result<ColumnRowStream> {
1188 self.stream_row_query(view_name, columns, where_sql, order_by, limit, offset)
1189 .await
1190 }
1191
1192 /// Retrieve query performance statistics from the database.
1193 ///
1194 /// Returns the top-N queries ordered by total execution time (descending).
1195 /// The exact data source depends on the backend:
1196 /// - PostgreSQL: `pg_stat_statements` (requires extension)
1197 /// - MySQL: `performance_schema.events_statements_summary_by_digest`
1198 /// - SQL Server: `sys.dm_exec_query_stats`
1199 /// - SQLite / Wire: empty (no stats available)
1200 ///
1201 /// # Arguments
1202 ///
1203 /// * `limit` - Maximum number of entries to return.
1204 ///
1205 /// # Errors
1206 ///
1207 /// Returns `FraiseQLError::Database` if the stats query fails.
1208 async fn query_stats(&self, _limit: u32) -> Result<Vec<crate::types::QueryStatEntry>> {
1209 Ok(vec![])
1210 }
1211
1212 /// Retrieve statistics for a single query by its ID.
1213 ///
1214 /// The default implementation fetches up to 1000 entries via
1215 /// [`query_stats`](Self::query_stats) and filters client-side.
1216 /// Backends with efficient single-query lookup (PostgreSQL, SQL Server)
1217 /// should override with a `WHERE` clause.
1218 ///
1219 /// # Errors
1220 ///
1221 /// Returns `FraiseQLError::Database` if the underlying query fails.
1222 async fn query_stats_by_id(&self, id: &str) -> Result<Option<crate::types::QueryStatEntry>> {
1223 let stats = self.query_stats(1000).await?;
1224 Ok(stats.into_iter().find(|e| e.query_id == id))
1225 }
1226
1227 /// Reset query performance statistics.
1228 ///
1229 /// Only PostgreSQL supports this (via `pg_stat_statements_reset()`).
1230 /// All other adapters return `Unsupported`.
1231 ///
1232 /// # Errors
1233 ///
1234 /// Returns `FraiseQLError::Unsupported` for adapters that cannot reset stats.
1235 /// Returns `FraiseQLError::Database` if the reset command fails.
1236 async fn reset_query_stats(&self) -> Result<()> {
1237 Err(FraiseQLError::Unsupported {
1238 message: "Query stats reset is not supported by this database adapter".to_string(),
1239 })
1240 }
1241
1242 /// Notify the adapter that the schema has changed.
1243 ///
1244 /// Called during hot-reload after the new schema has been validated.
1245 /// Adapters that maintain schema-dependent state (e.g. cache keyed by schema
1246 /// version) should clear or rebuild that state here.
1247 ///
1248 /// The default implementation is a no-op.
1249 fn on_schema_reload(&self) {}
1250
1251 /// Run one operator-supplied statement under the bounds in `request` (#962).
1252 ///
1253 /// This is the only method on this trait whose SQL FraiseQL did not generate.
1254 /// It exists for the Studio SQL console, and it is deliberately the *narrowest*
1255 /// possible shape for that: a single statement, in a transaction whose mode,
1256 /// timeout, row budget and session identity are all decided by the caller and
1257 /// enforced by the database. See [`AdminSqlRequest`] for what each bound buys.
1258 ///
1259 /// # Default implementation
1260 ///
1261 /// **Refuses.** An adapter that has not implemented containment does not get
1262 /// to run arbitrary SQL by inheriting a default that "just works" — the
1263 /// convenient default here would be `execute_raw_query`, which has no
1264 /// transaction, no timeout, no row cap and no way to roll back. Every
1265 /// adapter that should serve this endpoint says so by overriding, and the
1266 /// endpoint answers "unsupported" for the rest.
1267 ///
1268 /// A wrapping adapter must forward explicitly for the same reason it forwards
1269 /// [`stream_with_projection`](Self::stream_with_projection): inheriting the
1270 /// default here turns a capable adapter into an incapable one, and the only
1271 /// symptom is a refusal nobody asked for.
1272 ///
1273 /// # Errors
1274 ///
1275 /// Returns `FraiseQLError::Unsupported` by default. Implementations return
1276 /// `FraiseQLError::Database` for anything PostgreSQL rejects — which
1277 /// includes the write refused by a `READ ONLY` transaction (SQLSTATE
1278 /// `25006`) and the statement cancelled by the timeout (`57014`).
1279 async fn execute_admin_sql(&self, _request: &AdminSqlRequest) -> Result<AdminSqlOutcome> {
1280 Err(FraiseQLError::Unsupported {
1281 message: "Operator-supplied SQL execution is not supported by this database adapter."
1282 .to_string(),
1283 })
1284 }
1285}
1286
1287/// The `SELECT` behind [`DatabaseAdapter::execute_row_query`] and its streaming
1288/// sibling.
1289///
1290/// Shared rather than written twice: the two methods answer the same question and
1291/// a transport that got a different row set depending on whether it streamed would
1292/// be the hardest kind of bug to see. `where_sql` and `order_by` are compiler-
1293/// generated SQL fragments, as the method contract requires.
1294#[must_use]
1295pub fn build_row_query_sql(
1296 view_name: &str,
1297 where_sql: Option<&str>,
1298 order_by: Option<&str>,
1299 limit: Option<u32>,
1300 offset: Option<u32>,
1301) -> String {
1302 use std::fmt::Write as _;
1303
1304 let mut sql = format!("SELECT * FROM \"{view_name}\"");
1305 if let Some(w) = where_sql {
1306 sql.push_str(" WHERE ");
1307 sql.push_str(w);
1308 }
1309 if let Some(ob) = order_by {
1310 sql.push_str(" ORDER BY ");
1311 sql.push_str(ob);
1312 }
1313 if let Some(l) = limit {
1314 let _ = write!(sql, " LIMIT {l}");
1315 }
1316 if let Some(o) = offset {
1317 let _ = write!(sql, " OFFSET {o}");
1318 }
1319 sql
1320}
1321
1322/// Decode one JSON-shaped row into the declared column order.
1323///
1324/// A column the row does not carry decodes to [`crate::types::ColumnValue::Null`]
1325/// rather than shortening the row: the transport encodes positionally against
1326/// `columns`.
1327#[must_use]
1328pub fn row_to_column_values<S: std::hash::BuildHasher>(
1329 row: &std::collections::HashMap<String, serde_json::Value, S>,
1330 columns: &[crate::types::ColumnSpec],
1331) -> Vec<crate::types::ColumnValue> {
1332 use crate::types::ColumnValue;
1333
1334 columns
1335 .iter()
1336 .map(|col| {
1337 row.get(&col.name).map_or(ColumnValue::Null, |v| match v {
1338 serde_json::Value::Null => ColumnValue::Null,
1339 serde_json::Value::Bool(b) => ColumnValue::Boolean(*b),
1340 serde_json::Value::Number(n) => {
1341 if let Some(i) = n.as_i64() {
1342 ColumnValue::Int64(i)
1343 } else if let Some(f) = n.as_f64() {
1344 ColumnValue::Float64(f)
1345 } else {
1346 ColumnValue::Text(n.to_string())
1347 }
1348 },
1349 serde_json::Value::String(s) => ColumnValue::Text(s.clone()),
1350 other => ColumnValue::Json(other.to_string()),
1351 })
1352 })
1353 .collect()
1354}
1355
1356#[cfg(test)]
1357mod tests;