hyperdb_api/
prepared.rs

1// Copyright (c) 2026, Salesforce, Inc. All rights reserved.
2// SPDX-License-Identifier: Apache-2.0 OR MIT
3
4//! High-level prepared statements.
5//!
6//! [`PreparedStatement`] wraps [`hyperdb_api_core::client::OwnedPreparedStatement`]
7//! and integrates it with the rest of the hyperdb-api surface:
8//!
9//! - Returns [`Rowset`](crate::Rowset) from streaming executions, so
10//!   row decoding, schema capture, and `Row::get::<T>()` work exactly
11//!   the same way as with [`Connection::execute_query`](crate::Connection::execute_query).
12//! - `execute` / `fetch_one` / `fetch_optional` / `fetch_all` /
13//!   `fetch_scalar` mirror the same helpers on `Connection`.
14//! - `OwnedPreparedStatement` already auto-closes on Drop at the lower
15//!   layer, so this wrapper needs no additional cleanup logic.
16
17use std::sync::Arc;
18
19use hyperdb_api_core::client::OwnedPreparedStatement;
20use hyperdb_api_core::types::Oid;
21
22use crate::connection::Connection;
23use crate::error::{Error, Result};
24use crate::params::{ParamFormat, ToSqlParam};
25use crate::result::{ResultColumn, ResultSchema, Row, RowValue, Rowset};
26use crate::transport::Transport;
27
28/// A handle to a server-side prepared statement.
29///
30/// Construct via [`Connection::prepare`] or
31/// [`Connection::prepare_typed`]. Holding this type keeps the statement
32/// allocated on the server; it is released automatically when the handle
33/// is dropped.
34///
35/// # Reuse
36///
37/// A single `PreparedStatement` can be executed many times with different
38/// parameter values — the server caches the parsed plan. This is the
39/// primary reason to use prepared statements over
40/// [`Connection::query_params`] for loops over user input.
41#[derive(Debug)]
42pub struct PreparedStatement<'conn> {
43    connection: &'conn Connection,
44    inner: OwnedPreparedStatement,
45    schema: Arc<ResultSchema>,
46}
47
48impl<'conn> PreparedStatement<'conn> {
49    #[expect(
50        clippy::unnecessary_wraps,
51        reason = "signature retained for API symmetry / future fallibility; returning Result/Option keeps callers from breaking when the function later grows failure cases"
52    )]
53    pub(crate) fn new(
54        connection: &'conn Connection,
55        inner: OwnedPreparedStatement,
56    ) -> Result<Self> {
57        let schema = build_schema_from_columns(inner.columns());
58        Ok(Self {
59            connection,
60            inner,
61            schema: Arc::new(schema),
62        })
63    }
64
65    /// Returns the number of parameters the statement expects.
66    #[must_use]
67    pub fn param_count(&self) -> usize {
68        self.inner.param_count()
69    }
70
71    /// Returns the parameter type OIDs (as the server inferred or the
72    /// caller explicitly passed to [`Connection::prepare_typed`]).
73    #[must_use]
74    pub fn param_types(&self) -> &[Oid] {
75        self.inner.param_types()
76    }
77
78    /// Returns the result-column schema. Always available — it was
79    /// captured during the Parse/Describe at prepare time.
80    #[must_use]
81    pub fn schema(&self) -> &ResultSchema {
82        &self.schema
83    }
84
85    /// The original SQL text.
86    #[must_use]
87    pub fn sql(&self) -> &str {
88        self.inner.query()
89    }
90
91    /// Executes the statement and returns a streaming [`Rowset`].
92    ///
93    /// Memory stays bounded to one chunk regardless of result size —
94    /// the prepared-statement equivalent of
95    /// [`Connection::execute_query`].
96    ///
97    /// # Errors
98    ///
99    /// - Returns [`Error::FeatureNotSupported`] if the underlying [`Connection`] is on
100    ///   gRPC transport (prepared statements are TCP-only).
101    /// - Returns [`Error::Server`] if the server rejects `Bind` or
102    ///   `Execute` (type mismatch, runtime error while streaming).
103    /// - Returns [`Error::Io`] on transport-level I/O failures.
104    pub fn query(&self, params: &[&dyn ToSqlParam]) -> Result<Rowset<'conn>> {
105        let (encoded, formats) = encode_params(params);
106        let client = tcp_client(self.connection)?;
107        let stream = client.execute_streaming_with_formats(
108            &self.inner,
109            encoded,
110            &formats,
111            crate::result::DEFAULT_BINARY_CHUNK_SIZE,
112        )?;
113        Ok(Rowset::from_prepared(stream))
114    }
115
116    /// Executes the statement as a command (INSERT / UPDATE / DELETE /
117    /// DDL) and returns the affected-row count.
118    ///
119    /// # Errors
120    ///
121    /// - Returns [`Error::FeatureNotSupported`] on gRPC transport.
122    /// - Returns [`Error::Server`] if the server rejects `Bind` or
123    ///   `Execute`.
124    /// - Returns [`Error::Io`] on transport-level I/O failures.
125    pub fn execute(&self, params: &[&dyn ToSqlParam]) -> Result<u64> {
126        let (encoded, formats) = encode_params(params);
127        let client = tcp_client(self.connection)?;
128        Ok(client.execute_no_result_with_formats(&self.inner, encoded, &formats)?)
129    }
130
131    /// Fetches exactly one row; errors if the result is empty.
132    ///
133    /// # Errors
134    ///
135    /// - Returns the error from [`query`](Self::query).
136    /// - Returns [`Error::Conversion`] with message `"Query returned no rows"`
137    ///   if the result is empty.
138    pub fn fetch_one(&self, params: &[&dyn ToSqlParam]) -> Result<Row> {
139        self.query(params)?.require_first_row()
140    }
141
142    /// Fetches at most one row; returns `None` if the result is empty.
143    ///
144    /// # Errors
145    ///
146    /// Returns the error from [`query`](Self::query); an empty result
147    /// yields `Ok(None)`.
148    pub fn fetch_optional(&self, params: &[&dyn ToSqlParam]) -> Result<Option<Row>> {
149        self.query(params)?.first_row()
150    }
151
152    /// Fetches every row into a `Vec`.
153    ///
154    /// # Errors
155    ///
156    /// Returns the error from [`query`](Self::query), or a transport error
157    /// produced while draining every chunk.
158    pub fn fetch_all(&self, params: &[&dyn ToSqlParam]) -> Result<Vec<Row>> {
159        self.query(params)?.collect_rows()
160    }
161
162    /// Fetches a single non-NULL scalar; errors on empty / NULL.
163    ///
164    /// # Errors
165    ///
166    /// - Returns the error from [`query`](Self::query).
167    /// - Returns [`Error::Conversion`] with message `"Query returned no rows"`
168    ///   if the result is empty.
169    /// - Returns [`Error::Conversion`] with message `"Scalar query returned NULL"`
170    ///   if the first cell is SQL `NULL`.
171    pub fn fetch_scalar<T: RowValue>(&self, params: &[&dyn ToSqlParam]) -> Result<T> {
172        self.query(params)?.require_scalar()
173    }
174
175    /// Fetches a single scalar, allowing NULL as `None`.
176    ///
177    /// # Errors
178    ///
179    /// Returns the error from [`query`](Self::query). An empty result
180    /// still errors (see [`fetch_scalar`](Self::fetch_scalar)); SQL `NULL`
181    /// yields `Ok(None)`.
182    pub fn fetch_optional_scalar<T: RowValue>(
183        &self,
184        params: &[&dyn ToSqlParam],
185    ) -> Result<Option<T>> {
186        self.query(params)?.scalar()
187    }
188}
189
190/// Encode a slice of `&dyn ToSqlParam` into the wire bytes the
191/// prepared-statement Bind message expects, plus the matching per-parameter
192/// format code. `None` encodes SQL NULL.
193///
194/// The two vectors are always the same length; index `i` of the format vector
195/// describes index `i` of the byte vector.
196pub(crate) fn encode_params(
197    params: &[&dyn ToSqlParam],
198) -> (Vec<Option<Vec<u8>>>, Vec<ParamFormat>) {
199    params
200        .iter()
201        .map(|p| (p.encode_param(), p.param_format()))
202        .collect()
203}
204
205/// Extract the underlying sync TCP client or error with a clear message
206/// if the connection is on gRPC.
207pub(crate) fn tcp_client(connection: &Connection) -> Result<&hyperdb_api_core::client::Client> {
208    match connection.transport() {
209        Transport::Tcp(tcp) => Ok(&tcp.client),
210        Transport::Grpc(_) => Err(Error::feature_not_supported(
211            "prepared statements are not supported over gRPC transport",
212        )),
213    }
214}
215
216/// Build a `ResultSchema` from a slice of `hyperdb_api_core::client::Column`, using
217/// `SqlType::from_oid_and_modifier` so NUMERIC / VARCHAR modifiers are
218/// preserved.
219fn build_schema_from_columns(cols: &[hyperdb_api_core::client::Column]) -> ResultSchema {
220    let columns = cols
221        .iter()
222        .enumerate()
223        .map(|(idx, col)| {
224            let sql_type = hyperdb_api_core::types::SqlType::from_oid_and_modifier(
225                col.type_oid().0,
226                col.type_modifier(),
227            );
228            ResultColumn::new(col.name(), sql_type, idx)
229        })
230        .collect();
231    ResultSchema::from_columns(columns)
232}
hyperdb_api/prepared.rs

hyperdb_api/
prepared.rs
