Skip to main content

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, &param_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}