hyperdb_api/
async_prepared.rs

1// Copyright (c) 2026, Salesforce, Inc. All rights reserved.
2// SPDX-License-Identifier: Apache-2.0 OR MIT
3
4//! High-level async prepared statements.
5//!
6//! Async mirror of [`PreparedStatement`](crate::PreparedStatement); see
7//! that type's docs for the design rationale.
8
9use std::sync::Arc;
10
11use hyperdb_api_core::client::AsyncPreparedStatement as LowLevelAsyncPreparedStatement;
12use hyperdb_api_core::types::Oid;
13
14use crate::async_connection::AsyncConnection;
15use crate::async_result::AsyncRowset;
16use crate::async_transport::AsyncTransport;
17use crate::error::{Error, Result};
18use crate::params::{ParamFormat, ToSqlParam};
19use crate::result::{ResultColumn, ResultSchema, Row, RowValue};
20
21/// A handle to a server-side prepared statement (async).
22///
23/// Construct via [`AsyncConnection::prepare`] or
24/// [`AsyncConnection::prepare_typed`]. Holding this type keeps the
25/// statement allocated on the server; it is released automatically when
26/// the handle is dropped (best-effort — see
27/// [`hyperdb_api_core::client::AsyncPreparedStatement`] for the Drop semantics).
28/// The owned variant [`AsyncPreparedStatementOwned`] also provides an
29/// explicit [`close`](AsyncPreparedStatementOwned::close) method for
30/// callers that want deterministic cleanup.
31#[derive(Debug)]
32pub struct AsyncPreparedStatement<'conn> {
33    connection: &'conn AsyncConnection,
34    inner: LowLevelAsyncPreparedStatement,
35    schema: Arc<ResultSchema>,
36}
37
38impl<'conn> AsyncPreparedStatement<'conn> {
39    #[expect(
40        clippy::unnecessary_wraps,
41        reason = "signature retained for API symmetry / future fallibility; returning Result/Option keeps callers from breaking when the function later grows failure cases"
42    )]
43    pub(crate) fn new(
44        connection: &'conn AsyncConnection,
45        inner: LowLevelAsyncPreparedStatement,
46    ) -> Result<Self> {
47        let schema = build_schema_from_columns(inner.columns());
48        Ok(Self {
49            connection,
50            inner,
51            schema: Arc::new(schema),
52        })
53    }
54
55    /// Number of parameters the statement expects.
56    #[must_use]
57    pub fn param_count(&self) -> usize {
58        self.inner.param_count()
59    }
60
61    /// Parameter type OIDs.
62    #[must_use]
63    pub fn param_types(&self) -> &[Oid] {
64        self.inner.param_types()
65    }
66
67    /// Result-column schema, always available (captured at prepare time).
68    #[must_use]
69    pub fn schema(&self) -> &ResultSchema {
70        &self.schema
71    }
72
73    /// The original SQL text.
74    #[must_use]
75    pub fn sql(&self) -> &str {
76        self.inner.query()
77    }
78
79    /// Executes the statement and returns a streaming [`AsyncRowset`].
80    ///
81    /// # Errors
82    ///
83    /// - Returns [`Error::FeatureNotSupported`] on gRPC transport.
84    /// - Returns [`Error::Server`] if the server rejects `Bind` or
85    ///   `Execute`.
86    /// - Returns [`Error::Io`] on transport-level I/O failures.
87    pub async fn query(&self, params: &[&dyn ToSqlParam]) -> Result<AsyncRowset<'conn>> {
88        let (encoded, formats) = encode_params(params);
89        let client = async_tcp_client(self.connection)?;
90        let stream = client
91            .execute_prepared_streaming_with_formats(
92                &self.inner,
93                encoded,
94                &formats,
95                crate::result::DEFAULT_BINARY_CHUNK_SIZE,
96            )
97            .await?;
98        Ok(AsyncRowset::from_prepared(stream))
99    }
100
101    /// Executes the statement as a command and returns the affected-row
102    /// count (async).
103    ///
104    /// # Errors
105    ///
106    /// - Returns [`Error::FeatureNotSupported`] on gRPC transport.
107    /// - Returns [`Error::Server`] if the server rejects `Bind` or
108    ///   `Execute`.
109    /// - Returns [`Error::Io`] on transport-level I/O failures.
110    pub async fn execute(&self, params: &[&dyn ToSqlParam]) -> Result<u64> {
111        let (encoded, formats) = encode_params(params);
112        let client = async_tcp_client(self.connection)?;
113        Ok(client
114            .execute_prepared_no_result_with_formats(&self.inner, encoded, &formats)
115            .await?)
116    }
117
118    /// Fetches exactly one row; errors if the result is empty.
119    ///
120    /// # Errors
121    ///
122    /// - Returns the error from [`query`](Self::query).
123    /// - Returns [`Error::Conversion`] with message `"Query returned no rows"`
124    ///   if the result is empty.
125    pub async fn fetch_one(&self, params: &[&dyn ToSqlParam]) -> Result<Row> {
126        self.query(params).await?.require_first_row().await
127    }
128
129    /// Fetches at most one row; returns `None` if the result is empty.
130    ///
131    /// # Errors
132    ///
133    /// Returns the error from [`query`](Self::query); an empty result
134    /// yields `Ok(None)`.
135    pub async fn fetch_optional(&self, params: &[&dyn ToSqlParam]) -> Result<Option<Row>> {
136        self.query(params).await?.first_row().await
137    }
138
139    /// Fetches every row into a `Vec`.
140    ///
141    /// # Errors
142    ///
143    /// Returns the error from [`query`](Self::query), or a transport
144    /// error produced while draining every chunk.
145    pub async fn fetch_all(&self, params: &[&dyn ToSqlParam]) -> Result<Vec<Row>> {
146        self.query(params).await?.collect_rows().await
147    }
148
149    /// Fetches a single non-NULL scalar; errors on empty / NULL.
150    ///
151    /// # Errors
152    ///
153    /// - Returns the error from [`query`](Self::query).
154    /// - Returns [`Error::Conversion`] with message `"Query returned no rows"`
155    ///   if the result is empty.
156    /// - Returns [`Error::Conversion`] with message `"Scalar query returned NULL"`
157    ///   if the first cell is SQL `NULL`.
158    pub async fn fetch_scalar<T: RowValue>(&self, params: &[&dyn ToSqlParam]) -> Result<T> {
159        self.query(params).await?.require_scalar().await
160    }
161
162    /// Fetches a single scalar, allowing NULL as `None`.
163    ///
164    /// # Errors
165    ///
166    /// Returns the error from [`query`](Self::query); SQL `NULL` yields
167    /// `Ok(None)`.
168    pub async fn fetch_optional_scalar<T: RowValue>(
169        &self,
170        params: &[&dyn ToSqlParam],
171    ) -> Result<Option<T>> {
172        self.query(params).await?.scalar().await
173    }
174}
175
176/// Async twin of [`crate::prepared::encode_params`] — wire bytes plus the
177/// matching per-parameter format code, index for index.
178pub(crate) fn encode_params(
179    params: &[&dyn ToSqlParam],
180) -> (Vec<Option<Vec<u8>>>, Vec<ParamFormat>) {
181    params
182        .iter()
183        .map(|p| (p.encode_param(), p.param_format()))
184        .collect()
185}
186
187// =============================================================================
188// AsyncPreparedStatementOwned — lifetime-free variant
189// =============================================================================
190
191/// Owned-handle variant of [`AsyncPreparedStatement`] that holds an
192/// `Arc<AsyncConnection>` instead of a borrow.
193///
194/// Semantics are identical to [`AsyncPreparedStatement`]. The only
195/// difference is that this variant is `'static` and can therefore live
196/// in structs that can't carry lifetimes — N-API classes, `tokio::spawn`
197/// tasks that outlive the constructor, etc.
198#[derive(Debug)]
199pub struct AsyncPreparedStatementOwned {
200    connection: Arc<AsyncConnection>,
201    inner: LowLevelAsyncPreparedStatement,
202    schema: Arc<ResultSchema>,
203}
204
205impl AsyncPreparedStatementOwned {
206    #[expect(
207        clippy::unnecessary_wraps,
208        reason = "signature retained for API symmetry / future fallibility; returning Result/Option keeps callers from breaking when the function later grows failure cases"
209    )]
210    pub(crate) fn new(
211        connection: Arc<AsyncConnection>,
212        inner: LowLevelAsyncPreparedStatement,
213    ) -> Result<Self> {
214        let schema = build_schema_from_columns(inner.columns());
215        Ok(Self {
216            connection,
217            inner,
218            schema: Arc::new(schema),
219        })
220    }
221
222    /// Number of parameters the statement expects.
223    #[must_use]
224    pub fn param_count(&self) -> usize {
225        self.inner.param_count()
226    }
227
228    /// Parameter type OIDs.
229    #[must_use]
230    pub fn param_types(&self) -> &[Oid] {
231        self.inner.param_types()
232    }
233
234    /// Result-column schema, captured at prepare time.
235    #[must_use]
236    pub fn schema(&self) -> &ResultSchema {
237        &self.schema
238    }
239
240    /// Original SQL text.
241    #[must_use]
242    pub fn sql(&self) -> &str {
243        self.inner.query()
244    }
245
246    /// Executes the statement and returns a materialized `Vec<Row>`.
247    ///
248    /// Unlike [`AsyncPreparedStatement::query`], the owned variant
249    /// returns an owned `Vec<Row>` rather than a streaming
250    /// [`AsyncRowset`]: `AsyncRowset` is itself lifetime-bound to
251    /// the connection's mutex guard, which defeats the purpose of the
252    /// owned wrapper. N-API callers that want streaming should fall
253    /// back to the non-owned `AsyncPreparedStatement` via
254    /// [`AsyncConnection::prepare`] or use the non-streaming query
255    /// methods below.
256    ///
257    /// # Errors
258    ///
259    /// - Returns [`Error::FeatureNotSupported`] on gRPC transport.
260    /// - Returns [`Error::Server`] if the server rejects `Bind` or
261    ///   `Execute`, or raises a runtime error while streaming.
262    /// - Returns [`Error::Io`] on transport-level I/O failures.
263    pub async fn fetch_all(&self, params: &[&dyn ToSqlParam]) -> Result<Vec<Row>> {
264        let (encoded, formats) = encode_params(params);
265        let client = async_tcp_client_arc(&self.connection)?;
266        let stream = client
267            .execute_prepared_streaming_with_formats(
268                &self.inner,
269                encoded,
270                &formats,
271                crate::result::DEFAULT_BINARY_CHUNK_SIZE,
272            )
273            .await?;
274        let rowset = AsyncRowset::from_prepared(stream);
275        rowset.collect_rows().await
276    }
277
278    /// Executes the statement as a command; returns the affected-row count.
279    ///
280    /// # Errors
281    ///
282    /// - Returns [`Error::FeatureNotSupported`] on gRPC transport.
283    /// - Returns [`Error::Server`] if the server rejects `Bind` or
284    ///   `Execute`.
285    /// - Returns [`Error::Io`] on transport-level I/O failures.
286    pub async fn execute(&self, params: &[&dyn ToSqlParam]) -> Result<u64> {
287        let (encoded, formats) = encode_params(params);
288        let client = async_tcp_client_arc(&self.connection)?;
289        Ok(client
290            .execute_prepared_no_result_with_formats(&self.inner, encoded, &formats)
291            .await?)
292    }
293
294    /// Fetches exactly one row; errors on empty.
295    ///
296    /// # Errors
297    ///
298    /// - Returns the error from [`fetch_all`](Self::fetch_all).
299    /// - Returns [`Error::Conversion`] with message `"Query returned no rows"`
300    ///   if the result is empty.
301    pub async fn fetch_one(&self, params: &[&dyn ToSqlParam]) -> Result<Row> {
302        self.fetch_all(params)
303            .await?
304            .into_iter()
305            .next()
306            .ok_or_else(|| crate::error::Error::conversion("Query returned no rows"))
307    }
308
309    /// Fetches at most one row; `None` on empty.
310    ///
311    /// # Errors
312    ///
313    /// Returns the error from [`fetch_all`](Self::fetch_all); an empty
314    /// result yields `Ok(None)`.
315    pub async fn fetch_optional(&self, params: &[&dyn ToSqlParam]) -> Result<Option<Row>> {
316        Ok(self.fetch_all(params).await?.into_iter().next())
317    }
318
319    /// Fetches the first column of the first row as `T`.
320    ///
321    /// # Errors
322    ///
323    /// - Returns the error from [`fetch_one`](Self::fetch_one).
324    /// - Returns [`Error::Conversion`] with message `"Scalar query returned NULL"`
325    ///   if the first cell is SQL `NULL`.
326    pub async fn fetch_scalar<T: RowValue>(&self, params: &[&dyn ToSqlParam]) -> Result<T> {
327        let row = self.fetch_one(params).await?;
328        row.get::<T>(0)
329            .ok_or_else(|| crate::error::Error::conversion("Scalar query returned NULL"))
330    }
331
332    /// Fetches the first column of the first row as `Option<T>`.
333    ///
334    /// # Errors
335    ///
336    /// Returns the error from [`fetch_optional`](Self::fetch_optional);
337    /// SQL `NULL` yields `Ok(None)`.
338    pub async fn fetch_optional_scalar<T: RowValue>(
339        &self,
340        params: &[&dyn ToSqlParam],
341    ) -> Result<Option<T>> {
342        Ok(self
343            .fetch_optional(params)
344            .await?
345            .and_then(|r| r.get::<T>(0)))
346    }
347
348    /// Explicitly close the statement on the server.
349    ///
350    /// Equivalent to dropping the struct — the inner
351    /// `hyperdb_api_core::client::AsyncPreparedStatement` has its own Drop-time
352    /// best-effort close.
353    pub fn close(self) {
354        drop(self);
355    }
356}
357
358fn async_tcp_client_arc(
359    connection: &Arc<AsyncConnection>,
360) -> Result<&hyperdb_api_core::client::AsyncClient> {
361    match connection.transport() {
362        AsyncTransport::Tcp(tcp) => Ok(&tcp.client),
363        AsyncTransport::Grpc(_) => Err(Error::feature_not_supported(
364            "prepared statements are not supported over gRPC transport",
365        )),
366    }
367}
368
369pub(crate) fn async_tcp_client(
370    connection: &AsyncConnection,
371) -> Result<&hyperdb_api_core::client::AsyncClient> {
372    match connection.transport() {
373        AsyncTransport::Tcp(tcp) => Ok(&tcp.client),
374        AsyncTransport::Grpc(_) => Err(Error::feature_not_supported(
375            "prepared statements are not supported over gRPC transport",
376        )),
377    }
378}
379
380fn build_schema_from_columns(cols: &[hyperdb_api_core::client::Column]) -> ResultSchema {
381    let columns = cols
382        .iter()
383        .enumerate()
384        .map(|(idx, col)| {
385            let sql_type = hyperdb_api_core::types::SqlType::from_oid_and_modifier(
386                col.type_oid().0,
387                col.type_modifier(),
388            );
389            ResultColumn::new(col.name(), sql_type, idx)
390        })
391        .collect();
392    ResultSchema::from_columns(columns)
393}
hyperdb_api/async_prepared.rs

hyperdb_api/
async_prepared.rs
