fraiseql_db/postgres/adapter/mod.rs
1//! PostgreSQL database adapter implementation.
2
3mod database;
4mod query_stats;
5mod relay;
6
7#[cfg(test)]
8mod tests;
9
10#[cfg(all(test, feature = "test-postgres"))]
11mod integration_tests;
12
13use std::{fmt::Write, time::Duration};
14
15use deadpool_postgres::{Config, ManagerConfig, Pool, RecyclingMethod, Runtime};
16use fraiseql_error::{FraiseQLError, Result};
17use tokio_postgres::{NoTls, Row};
18
19use super::where_generator::PostgresWhereGenerator;
20use crate::{
21 dialect::PostgresDialect,
22 identifier::quote_postgres_identifier,
23 order_by::append_order_by,
24 traits::DatabaseAdapter,
25 types::{
26 DatabaseType, JsonbValue, QueryParam,
27 sql_hints::{OrderByClause, SqlProjectionHint},
28 },
29 where_clause::WhereClause,
30};
31
32/// Extract the JSONB `data` cell from a result row, failing loud rather than
33/// panicking.
34///
35/// `Row::get` panics on SQL NULL or a non-JSONB column type, so a backing view
36/// that projects NULL `data` (e.g. via a LEFT JOIN) or a mistyped `data` column
37/// turned a query into a request-path panic — the PostgreSQL adapter was the
38/// only backend that aborted here instead of returning an error (audit H34).
39/// Both the NULL and the type-mismatch cases now map to
40/// [`FraiseQLError::Database`], including a bounded, char-safe slice of the
41/// query so an operator can identify the offending view.
42fn jsonb_cell<I>(row: &Row, column: I, sql: &str) -> Result<JsonbValue>
43where
44 I: tokio_postgres::row::RowIndex + std::fmt::Display,
45{
46 // `.chars().take(..)` is inherently char-boundary-safe (no byte slicing).
47 let query_preview = || sql.chars().take(200).collect::<String>();
48 match row.try_get::<_, Option<serde_json::Value>>(column) {
49 Ok(Some(value)) => Ok(JsonbValue::new(value)),
50 Ok(None) => Err(FraiseQLError::Database {
51 message: format!(
52 "Query returned a NULL `data` column; the backing view must project a \
53 non-NULL JSONB `data` value (a view yielding NULL `data`, e.g. via a \
54 LEFT JOIN, is unsupported). Query: {}",
55 query_preview()
56 ),
57 sql_state: None,
58 }),
59 Err(e) => Err(FraiseQLError::Database {
60 message: format!(
61 "Failed to read the `data` column as JSONB ({e}); the backing view must \
62 project a JSONB `data` column. Query: {}",
63 query_preview()
64 ),
65 sql_state: None,
66 }),
67 }
68}
69
70/// Default maximum pool size for PostgreSQL connections.
71/// Increased from 10 to 25 to prevent pool exhaustion under concurrent
72/// nested query load (fixes Issue #41).
73const DEFAULT_POOL_SIZE: usize = 25;
74
75/// Maximum retries for connection acquisition with exponential backoff.
76const MAX_CONNECTION_RETRIES: u32 = 3;
77
78/// Base delay in milliseconds for connection retry backoff.
79const CONNECTION_RETRY_DELAY_MS: u64 = 50;
80
81/// Configuration for connection pool construction and pre-warming.
82///
83/// Controls the minimum guaranteed connections (pre-warmed at startup),
84/// the maximum pool ceiling, and the wait/create timeout for connection
85/// acquisition.
86///
87/// # Example
88///
89/// ```rust
90/// use fraiseql_db::postgres::PoolPrewarmConfig;
91///
92/// let cfg = PoolPrewarmConfig {
93/// min_size: 5,
94/// max_size: 20,
95/// timeout_secs: Some(30),
96/// };
97/// ```
98#[derive(Debug, Clone)]
99pub struct PoolPrewarmConfig {
100 /// Number of connections to establish at pool creation time.
101 ///
102 /// After the pool is created, `min_size` connections are opened eagerly
103 /// so they are ready when the first request arrives. Set to `0` to disable
104 /// pre-warming (lazy init — one connection from the startup health check).
105 pub min_size: usize,
106
107 /// Maximum number of connections the pool may hold.
108 pub max_size: usize,
109
110 /// Optional timeout (in seconds) for connection acquisition and creation.
111 ///
112 /// Applied to both the `wait` (blocked waiting for an idle connection) and
113 /// `create` (time to open a new TCP connection to PostgreSQL) deadpool slots.
114 /// When `None`, acquisition can block indefinitely on pool exhaustion.
115 pub timeout_secs: Option<u64>,
116}
117
118/// Build a `deadpool-postgres` pool with an optional wait/create timeout.
119///
120/// # Errors
121///
122/// Returns `FraiseQLError::ConnectionPool` if pool creation fails (e.g., unparseable URL).
123fn build_pool(connection_string: &str, max_size: usize, timeout_secs: Option<u64>) -> Result<Pool> {
124 let mut cfg = Config::new();
125 cfg.url = Some(connection_string.to_string());
126 cfg.manager = Some(ManagerConfig {
127 recycling_method: RecyclingMethod::Fast,
128 });
129
130 let mut pool_cfg = deadpool_postgres::PoolConfig::new(max_size);
131 if let Some(secs) = timeout_secs {
132 let t = Duration::from_secs(secs);
133 pool_cfg.timeouts.wait = Some(t);
134 pool_cfg.timeouts.create = Some(t);
135 // `recycle` intentionally stays None — fast recycle, not user-configurable.
136 }
137 cfg.pool = Some(pool_cfg);
138
139 cfg.create_pool(Some(Runtime::Tokio1), NoTls)
140 .map_err(|e| FraiseQLError::ConnectionPool {
141 message: format!("Failed to create connection pool: {e}"),
142 })
143}
144
145/// Escape a JSONB key for use in a PostgreSQL string literal (`data->>'key'`).
146///
147/// PostgreSQL string literals use single-quote doubling for escaping (`'` → `''`).
148/// This function is defense-in-depth: `OrderByClause` already rejects field names
149/// that are not valid GraphQL identifiers (which cannot contain `'`), but this
150/// escaping ensures correctness for any future caller that bypasses that validation.
151pub(super) fn escape_jsonb_key(key: &str) -> String {
152 key.replace('\'', "''")
153}
154
155/// PostgreSQL database adapter with connection pooling.
156///
157/// Uses `deadpool-postgres` for connection pooling and `tokio-postgres` for async queries.
158///
159/// # Example
160///
161/// ```rust,no_run
162/// use fraiseql_db::postgres::PostgresAdapter;
163/// use fraiseql_db::{DatabaseAdapter, WhereClause, WhereOperator};
164/// use serde_json::json;
165///
166/// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
167/// // Create adapter with connection string
168/// let adapter = PostgresAdapter::new("postgresql://localhost/mydb").await?;
169///
170/// // Execute query
171/// let where_clause = WhereClause::Field {
172/// path: vec!["email".to_string()],
173/// operator: WhereOperator::Icontains,
174/// value: json!("example.com"),
175/// };
176///
177/// let results = adapter
178/// .execute_where_query("v_user", Some(&where_clause), Some(10), None, None)
179/// .await?;
180///
181/// println!("Found {} users", results.len());
182/// # Ok(())
183/// # }
184/// ```
185#[derive(Clone)]
186pub struct PostgresAdapter {
187 pub(super) pool: Pool,
188 /// Whether mutation timing injection is enabled.
189 mutation_timing_enabled: bool,
190 /// The PostgreSQL session variable name for timing.
191 timing_variable_name: String,
192}
193
194impl std::fmt::Debug for PostgresAdapter {
195 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
196 f.debug_struct("PostgresAdapter")
197 .field("mutation_timing_enabled", &self.mutation_timing_enabled)
198 .field("timing_variable_name", &self.timing_variable_name)
199 .field("pool", &"<Pool>")
200 .finish()
201 }
202}
203
204impl PostgresAdapter {
205 /// Create new PostgreSQL adapter with default pool configuration.
206 ///
207 /// # Arguments
208 ///
209 /// * `connection_string` - PostgreSQL connection string (e.g., "postgresql://localhost/mydb")
210 ///
211 /// # Errors
212 ///
213 /// Returns `FraiseQLError::ConnectionPool` if pool creation fails.
214 ///
215 /// # Example
216 ///
217 /// ```rust,no_run
218 /// # use fraiseql_db::postgres::PostgresAdapter;
219 /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
220 /// let adapter = PostgresAdapter::new("postgresql://localhost/mydb").await?;
221 /// # Ok(())
222 /// # }
223 /// ```
224 pub async fn new(connection_string: &str) -> Result<Self> {
225 Self::with_pool_config(
226 connection_string,
227 PoolPrewarmConfig {
228 min_size: 0,
229 max_size: DEFAULT_POOL_SIZE,
230 timeout_secs: None,
231 },
232 )
233 .await
234 }
235
236 /// Create new PostgreSQL adapter with pre-warming and timeout configuration.
237 ///
238 /// Constructs the pool, runs a startup health check, then eagerly opens
239 /// `cfg.min_size` connections so they are ready when the first request arrives.
240 ///
241 /// # Arguments
242 ///
243 /// * `connection_string` - PostgreSQL connection string
244 /// * `cfg` - Pool pre-warming and timeout configuration
245 ///
246 /// # Errors
247 ///
248 /// Returns `FraiseQLError::ConnectionPool` if pool creation or the startup
249 /// health check fails.
250 pub async fn with_pool_config(connection_string: &str, cfg: PoolPrewarmConfig) -> Result<Self> {
251 let pool = build_pool(connection_string, cfg.max_size, cfg.timeout_secs)?;
252
253 // Startup health check — establishes the first connection.
254 let client = pool.get().await.map_err(|e| FraiseQLError::ConnectionPool {
255 message: format!("Failed to acquire connection: {e}"),
256 })?;
257
258 client.query("SELECT 1", &[]).await.map_err(|e| FraiseQLError::Database {
259 message: format!("Failed to connect to database: {e}"),
260 sql_state: e.code().map(|c| c.code().to_string()),
261 })?;
262
263 // Drop client back to the pool before pre-warming so that the health-check
264 // connection counts as idle slot #1.
265 drop(client);
266
267 let adapter = Self {
268 pool,
269 mutation_timing_enabled: false,
270 timing_variable_name: "fraiseql.started_at".to_string(),
271 };
272
273 // Pre-warm: open `min_size - 1` additional connections (one already exists).
274 let warm_target = cfg.min_size.min(cfg.max_size).saturating_sub(1);
275 if warm_target > 0 {
276 adapter.prewarm(warm_target).await;
277 }
278
279 Ok(adapter)
280 }
281
282 /// Create new PostgreSQL adapter with custom pool size.
283 ///
284 /// # Arguments
285 ///
286 /// * `connection_string` - PostgreSQL connection string
287 /// * `max_size` - Maximum number of connections in pool
288 ///
289 /// # Errors
290 ///
291 /// Returns `FraiseQLError::ConnectionPool` if pool creation fails.
292 pub async fn with_pool_size(connection_string: &str, max_size: usize) -> Result<Self> {
293 Self::with_pool_config(
294 connection_string,
295 PoolPrewarmConfig {
296 min_size: 0,
297 max_size,
298 timeout_secs: None,
299 },
300 )
301 .await
302 }
303
304 /// Pre-warm the pool by opening `count` additional connections.
305 ///
306 /// Pre-warming is best-effort: failures from individual connections are logged
307 /// but do not prevent startup. A 10-second outer timeout ensures the server
308 /// never blocks indefinitely on a slow or unreachable PostgreSQL instance.
309 async fn prewarm(&self, count: usize) {
310 use futures::future::join_all;
311 use tokio::time::timeout;
312
313 let handles: Vec<_> = (0..count)
314 .map(|_| {
315 let pool = self.pool.clone();
316 tokio::spawn(async move { pool.get().await })
317 })
318 .collect();
319
320 let result = timeout(Duration::from_secs(10), join_all(handles)).await;
321
322 let (succeeded, failed) = match result {
323 Ok(outcomes) => {
324 let s = outcomes
325 .iter()
326 .filter(|r| r.as_ref().map(|inner| inner.is_ok()).unwrap_or(false))
327 .count();
328 (s, count - s)
329 },
330 Err(_elapsed) => {
331 tracing::warn!(
332 target_connections = count,
333 "Pool pre-warm timed out after 10s; server will continue with partial pre-warm"
334 );
335 (0, count)
336 },
337 };
338
339 if failed > 0 {
340 tracing::warn!(
341 succeeded,
342 failed,
343 "Pool pre-warm: some connections could not be established"
344 );
345 } else {
346 tracing::info!(
347 idle_connections = succeeded + 1,
348 "PostgreSQL pool pre-warmed successfully"
349 );
350 }
351 }
352
353 /// Get a reference to the internal connection pool.
354 ///
355 /// This allows sharing the pool with other components like `PostgresIntrospector`.
356 #[must_use]
357 pub const fn pool(&self) -> &Pool {
358 &self.pool
359 }
360
361 /// Enable mutation timing injection.
362 ///
363 /// When enabled, `execute_function_call` wraps each mutation in a transaction
364 /// and sets a session variable to `clock_timestamp()::text` before execution,
365 /// allowing SQL functions to compute their own duration.
366 ///
367 /// # Arguments
368 ///
369 /// * `variable_name` - The PostgreSQL session variable name (e.g., `"fraiseql.started_at"`)
370 #[must_use]
371 pub fn with_mutation_timing(mut self, variable_name: &str) -> Self {
372 self.mutation_timing_enabled = true;
373 self.timing_variable_name = variable_name.to_string();
374 self
375 }
376
377 /// Returns whether mutation timing injection is enabled.
378 #[must_use]
379 pub const fn mutation_timing_enabled(&self) -> bool {
380 self.mutation_timing_enabled
381 }
382
383 /// Execute raw SQL query and return JSONB rows.
384 ///
385 /// # Errors
386 ///
387 /// Returns `FraiseQLError::Database` on query execution failure.
388 pub(super) async fn execute_raw(
389 &self,
390 sql: &str,
391 params: &[&(dyn tokio_postgres::types::ToSql + Sync)],
392 ) -> Result<Vec<JsonbValue>> {
393 let client = self.acquire_connection_with_retry().await?;
394
395 let rows: Vec<Row> =
396 client.query(sql, params).await.map_err(|e| FraiseQLError::Database {
397 message: format!("Query execution failed: {e}"),
398 sql_state: e.code().map(|c| c.code().to_string()),
399 })?;
400
401 let results = rows
402 .into_iter()
403 .map(|row| jsonb_cell(&row, 0, sql))
404 .collect::<Result<Vec<_>>>()?;
405
406 Ok(results)
407 }
408
409 /// Like [`execute_raw`](Self::execute_raw) but applies transaction-local
410 /// session variables on the same connection / transaction that runs the
411 /// query.
412 ///
413 /// `set_config(..., true)` and the `SELECT` share one transaction, so
414 /// PostgreSQL RLS policies backed by `current_setting()` see the configured
415 /// values (fixes #329).
416 ///
417 /// # Errors
418 ///
419 /// Returns `FraiseQLError::Database` on transaction, `set_config`, query, or
420 /// commit failure.
421 pub(super) async fn execute_raw_with_session(
422 &self,
423 sql: &str,
424 params: &[&(dyn tokio_postgres::types::ToSql + Sync)],
425 session_vars: &[(&str, &str)],
426 ) -> Result<Vec<JsonbValue>> {
427 let mut client = self.acquire_connection_with_retry().await?;
428 let txn =
429 client.build_transaction().start().await.map_err(|e| FraiseQLError::Database {
430 message: format!("Failed to start session-var transaction: {e}"),
431 sql_state: e.code().map(|c| c.code().to_string()),
432 })?;
433
434 database::apply_session_vars(&txn, session_vars).await?;
435
436 let rows: Vec<Row> = txn.query(sql, params).await.map_err(|e| FraiseQLError::Database {
437 message: format!("Query execution failed: {e}"),
438 sql_state: e.code().map(|c| c.code().to_string()),
439 })?;
440
441 txn.commit().await.map_err(|e| FraiseQLError::Database {
442 message: format!("Failed to commit session-var transaction: {e}"),
443 sql_state: e.code().map(|c| c.code().to_string()),
444 })?;
445
446 rows.into_iter().map(|row| jsonb_cell(&row, 0, sql)).collect()
447 }
448
449 /// Acquire a connection from the pool with retry logic.
450 ///
451 /// - `PoolError::Timeout`: the pool was exhausted for the full configured wait period. This is
452 /// not transient — retrying would only multiply the wait. Fails immediately.
453 /// - `PoolError::Backend` / create errors: potentially transient. Retries with exponential
454 /// backoff (up to `MAX_CONNECTION_RETRIES` attempts).
455 ///
456 /// # Errors
457 ///
458 /// Returns `FraiseQLError::ConnectionPool` on timeout or when all retries are exhausted.
459 pub(super) async fn acquire_connection_with_retry(&self) -> Result<deadpool_postgres::Client> {
460 use deadpool_postgres::PoolError;
461
462 let mut last_error = None;
463
464 for attempt in 0..MAX_CONNECTION_RETRIES {
465 match self.pool.get().await {
466 Ok(client) => {
467 if attempt > 0 {
468 tracing::info!(attempt, "Successfully acquired connection after retries");
469 }
470 return Ok(client);
471 },
472 // Pool exhausted for the full wait period — not transient, fail immediately.
473 Err(PoolError::Timeout(_)) => {
474 let metrics = self.pool_metrics();
475 tracing::error!(
476 available = metrics.idle_connections,
477 active = metrics.active_connections,
478 max = metrics.total_connections,
479 "Connection pool timeout: all connections busy"
480 );
481 return Err(FraiseQLError::ConnectionPool {
482 message: format!(
483 "Connection pool timeout: {}/{} connections busy. \
484 Increase pool_max_size or reduce concurrent load.",
485 metrics.active_connections, metrics.total_connections,
486 ),
487 });
488 },
489 // Backend/create errors are potentially transient — retry with backoff.
490 Err(e) => {
491 last_error = Some(e);
492 if attempt < MAX_CONNECTION_RETRIES - 1 {
493 let delay = CONNECTION_RETRY_DELAY_MS * (u64::from(attempt) + 1);
494 tracing::warn!(
495 attempt = attempt + 1,
496 total = MAX_CONNECTION_RETRIES,
497 delay_ms = delay,
498 "Transient connection error, retrying"
499 );
500 tokio::time::sleep(Duration::from_millis(delay)).await;
501 }
502 },
503 }
504 }
505
506 // All retries for transient errors exhausted.
507 let pool_metrics = self.pool_metrics();
508 tracing::error!(
509 retries = MAX_CONNECTION_RETRIES,
510 available = pool_metrics.idle_connections,
511 active = pool_metrics.active_connections,
512 max = pool_metrics.total_connections,
513 "Failed to acquire connection after all retries"
514 );
515
516 Err(FraiseQLError::ConnectionPool {
517 message: format!(
518 "Failed to acquire connection after {} retries: {}. \
519 Pool state: idle={}, active={}, max={}",
520 MAX_CONNECTION_RETRIES,
521 last_error.expect("last_error is set on every retry iteration"),
522 pool_metrics.idle_connections,
523 pool_metrics.active_connections,
524 pool_metrics.total_connections,
525 ),
526 })
527 }
528
529 /// Execute query with SQL field projection optimization.
530 ///
531 /// Uses the provided `SqlProjectionHint` to generate optimized SQL that projects
532 /// only the requested fields from the JSONB column, reducing network payload and
533 /// JSON deserialization overhead.
534 ///
535 /// # Arguments
536 ///
537 /// * `view` - View/table name to query
538 /// * `projection` - Optional SQL projection hint with field list
539 /// * `where_clause` - Optional WHERE clause for filtering
540 /// * `limit` - Optional row limit
541 ///
542 /// # Returns
543 ///
544 /// Vector of projected JSONB rows with only the requested fields
545 ///
546 /// # Errors
547 ///
548 /// Returns `FraiseQLError::Database` on query execution failure.
549 ///
550 /// # Panics
551 ///
552 /// Cannot panic in practice: the inner `expect` is guarded by an `is_none()` check
553 /// immediately above it.
554 ///
555 /// # Example
556 ///
557 /// ```no_run
558 /// // Requires: running PostgreSQL database.
559 /// use fraiseql_db::postgres::PostgresAdapter;
560 /// use fraiseql_db::types::SqlProjectionHint;
561 /// use fraiseql_db::DatabaseType;
562 ///
563 /// # async fn example(adapter: &PostgresAdapter) -> Result<(), Box<dyn std::error::Error>> {
564 /// let projection = SqlProjectionHint::new(
565 /// DatabaseType::PostgreSQL,
566 /// "jsonb_build_object('id', data->>'id')".to_string(),
567 /// 75,
568 /// );
569 ///
570 /// let results = adapter
571 /// .execute_with_projection("v_user", Some(&projection), None, Some(10), None)
572 /// .await?;
573 /// # Ok(())
574 /// # }
575 /// ```
576 /// Implementation of `execute_with_projection` with ORDER BY support.
577 ///
578 /// Called by both the inherent convenience method and the `DatabaseAdapter`
579 /// trait implementation.
580 pub(super) async fn execute_with_projection_impl(
581 &self,
582 view: &str,
583 projection: Option<&SqlProjectionHint>,
584 where_clause: Option<&WhereClause>,
585 limit: Option<u32>,
586 offset: Option<u32>,
587 order_by: Option<&[OrderByClause]>,
588 ) -> Result<Vec<JsonbValue>> {
589 // If no projection, fall back to standard query
590 if projection.is_none() {
591 return self.execute_where_query(view, where_clause, limit, offset, order_by).await;
592 }
593
594 let projection = projection.expect("projection is Some; None was returned above");
595
596 let (sql, typed_params) =
597 build_projection_select_sql(projection, view, where_clause, limit, offset, order_by)?;
598
599 tracing::debug!("SQL with projection = {}", sql);
600 tracing::debug!("typed_params = {:?}", typed_params);
601
602 let param_refs = crate::types::as_sql_param_refs(&typed_params);
603
604 self.execute_raw(&sql, ¶m_refs).await
605 }
606
607 /// Execute query with SQL field projection optimization.
608 ///
609 /// Convenience wrapper for callers that don't need ORDER BY.
610 /// See `execute_with_projection_impl` for details.
611 ///
612 /// # Errors
613 ///
614 /// Returns `FraiseQLError::Database` on query execution failure.
615 pub async fn execute_with_projection(
616 &self,
617 view: &str,
618 projection: Option<&SqlProjectionHint>,
619 where_clause: Option<&WhereClause>,
620 limit: Option<u32>,
621 offset: Option<u32>,
622 ) -> Result<Vec<JsonbValue>> {
623 self.execute_with_projection_impl(view, projection, where_clause, limit, offset, None)
624 .await
625 }
626}
627
628/// Build a parameterized `SELECT data FROM {view}` SQL string.
629///
630/// Shared by [`PostgresAdapter::execute_where_query`] and
631/// [`PostgresAdapter::explain_where_query`] so that SQL construction
632/// logic is never duplicated.
633///
634/// # Returns
635///
636/// `(sql, typed_params)` — the SQL string and the bound parameter values.
637///
638/// # Errors
639///
640/// Returns `FraiseQLError` if WHERE clause generation fails.
641pub(super) fn build_where_select_sql(
642 view: &str,
643 where_clause: Option<&WhereClause>,
644 limit: Option<u32>,
645 offset: Option<u32>,
646) -> Result<(String, Vec<QueryParam>)> {
647 build_where_select_sql_ordered(view, where_clause, limit, offset, None)
648}
649
650/// Build a parameterized `SELECT data FROM {view}` SQL string with optional ORDER BY.
651///
652/// ORDER BY is inserted between the WHERE clause and LIMIT/OFFSET as required by SQL.
653///
654/// # Returns
655///
656/// `(sql, typed_params)` — the SQL string and the bound parameter values.
657///
658/// # Errors
659///
660/// Returns `FraiseQLError` if WHERE clause generation or field name validation fails.
661pub(super) fn build_where_select_sql_ordered(
662 view: &str,
663 where_clause: Option<&WhereClause>,
664 limit: Option<u32>,
665 offset: Option<u32>,
666 order_by: Option<&[OrderByClause]>,
667) -> Result<(String, Vec<QueryParam>)> {
668 // Build base query
669 let mut sql = format!("SELECT data FROM {}", quote_postgres_identifier(view));
670
671 // Collect WHERE clause params (if any)
672 let mut typed_params: Vec<QueryParam> = if let Some(clause) = where_clause {
673 let generator = PostgresWhereGenerator::new(PostgresDialect);
674 let (where_sql, where_params) = generator.generate(clause)?;
675 sql.push_str(" WHERE ");
676 sql.push_str(&where_sql);
677
678 // Convert WHERE clause JSON values to QueryParam
679 where_params.into_iter().map(QueryParam::from).collect()
680 } else {
681 Vec::new()
682 };
683 let mut param_count = typed_params.len();
684
685 // ORDER BY must come before LIMIT/OFFSET in SQL.
686 append_order_by(&mut sql, order_by, DatabaseType::PostgreSQL)?;
687
688 // Add LIMIT as BigInt (PostgreSQL requires integer type for LIMIT).
689 // Reason (expect below): fmt::Write for String is infallible.
690 if let Some(lim) = limit {
691 param_count += 1;
692 write!(sql, " LIMIT ${param_count}").expect("write to String");
693 typed_params.push(QueryParam::BigInt(i64::from(lim)));
694 }
695
696 // Add OFFSET as BigInt (PostgreSQL requires integer type for OFFSET)
697 if let Some(off) = offset {
698 param_count += 1;
699 write!(sql, " OFFSET ${param_count}").expect("write to String");
700 typed_params.push(QueryParam::BigInt(i64::from(off)));
701 }
702
703 Ok((sql, typed_params))
704}
705
706/// Build a parameterized projection `SELECT` SQL string.
707///
708/// Mirrors the SQL produced inline by [`PostgresAdapter::execute_with_projection_impl`],
709/// extracted so the connection-affine `*_with_session` path can reuse it
710/// without acquiring its own connection.
711///
712/// # Returns
713///
714/// `(sql, typed_params)` — the SQL string and the bound parameter values.
715///
716/// # Errors
717///
718/// Returns `FraiseQLError` if WHERE clause generation fails.
719pub(super) fn build_projection_select_sql(
720 projection: &SqlProjectionHint,
721 view: &str,
722 where_clause: Option<&WhereClause>,
723 limit: Option<u32>,
724 offset: Option<u32>,
725 order_by: Option<&[OrderByClause]>,
726) -> Result<(String, Vec<QueryParam>)> {
727 // The projection_template is the SELECT clause with projection SQL,
728 // e.g. "jsonb_build_object('id', data->>'id', 'email', data->>'email')".
729 let mut sql = format!(
730 "SELECT {} FROM {}",
731 projection.projection_template,
732 quote_postgres_identifier(view)
733 );
734
735 let mut typed_params: Vec<QueryParam> = if let Some(clause) = where_clause {
736 let generator = PostgresWhereGenerator::new(PostgresDialect);
737 let (where_sql, where_params) = generator.generate(clause)?;
738 sql.push_str(" WHERE ");
739 sql.push_str(&where_sql);
740 where_params.into_iter().map(QueryParam::from).collect()
741 } else {
742 Vec::new()
743 };
744 let mut param_count = typed_params.len();
745
746 // ORDER BY must come before LIMIT/OFFSET in SQL.
747 append_order_by(&mut sql, order_by, DatabaseType::PostgreSQL)?;
748
749 // Append LIMIT/OFFSET as BigInt (PostgreSQL requires integer type).
750 // Reason (expect below): fmt::Write for String is infallible.
751 if let Some(lim) = limit {
752 param_count += 1;
753 write!(sql, " LIMIT ${param_count}").expect("write to String");
754 typed_params.push(QueryParam::BigInt(i64::from(lim)));
755 }
756
757 if let Some(off) = offset {
758 param_count += 1;
759 write!(sql, " OFFSET ${param_count}").expect("write to String");
760 typed_params.push(QueryParam::BigInt(i64::from(off)));
761 }
762
763 Ok((sql, typed_params))
764}