Skip to content

Commit c576945

Browse files
StefanSteinerStefanSteiner
authored
fix(api): parameterized FromRow methods (fetch_*_as_params / stream_as_params) (#152)
* fix(api): add sync parameterized FromRow methods (#137) Add fetch_one_as_params, fetch_all_as_params, and stream_as_params on Connection — the intersection of FromRow struct-mapping and ToSqlParam parameter binding. Each delegates to query_params (binary Parse/Bind/ Execute) then maps rows exactly as the existing *_as methods do; the prepared-statement guard rides along on the returned Rowset, so Drop ordering is preserved. Round-trip tests cover single/all/stream, multi-param binding, and the empty-result error path. * fix(api): add async parameterized FromRow methods (#137) Mirror the sync fetch_one_as_params / fetch_all_as_params / stream_as_params on AsyncConnection. fetch_one/fetch_all delegate to query_params; stream_as_params encodes params before entering the try_stream! generator (it can't borrow &[&dyn ToSqlParam] across await points) and rebuilds query_params' prepare/execute sequence inline, carrying the statement guard. The Prepared path exposes its schema at prepare time, so the column-index map is built once up front. gRPC FeatureNotSupported surfaces as the stream's first yielded item (documented), not eagerly. Async round-trip tests cover all three. * docs(api): document parameterized FromRow methods (#137) - FromRow and ToSqlParam rustdoc cross-reference the new _as_params trio. - row_mapping_forms example gains Form 6 (fetch_all_as_params + stream_as_params), wired into main and verified end-to-end. - docs/ROW_MAPPING.md: new 'Form 6 — Parameterized struct mapping' section + Choosing-a-form row; retitled 'Six Forms'. - README: note + snippet under Parameterized Queries. - Fix stale async query_params rustdoc that described text-mode escaping; it has done binary Parse/Bind/Execute since the sync parity work. * docs: fix stale 'five forms' line in ROW_MAPPING (#137) Final-sweep reviewer caught line 9 still saying 'All five forms' after the doc was retitled to Six Forms.
1 parent 5160975 commit c576945

9 files changed

Lines changed: 657 additions & 12 deletions

File tree

‎docs/ROW_MAPPING.md‎

Lines changed: 58 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,12 @@
1-
# Row Mapping: Five Forms
1+
# Row Mapping: Six Forms
22

3-
When querying Hyper, there are five ways to map result rows into Rust values —
4-
from fully manual to fully automatic. Forms 1–4 trade manual control for
3+
When querying Hyper, there are several ways to map result rows into Rust values
4+
— from fully manual to fully automatic. Forms 1–4 trade manual control for
55
convenience; Form 5 combines the automatic struct mapping of Form 4 with the
6-
constant-memory streaming of Form 1. Start with the simplest that fits your
7-
situation.
6+
constant-memory streaming of Form 1; Form 6 adds `$1` parameter binding to the
7+
struct mapping. Start with the simplest that fits your situation.
88

9-
All five forms are demonstrated end-to-end in one runnable example:
9+
All six forms are demonstrated end-to-end in one runnable example:
1010

1111
```
1212
cargo run -p hyperdb-api --example row_mapping_forms
@@ -307,6 +307,57 @@ async fn print_products(conn: &hyperdb_api::AsyncConnection) -> hyperdb_api::Res
307307

308308
---
309309

310+
## Form 6 — Parameterized struct mapping
311+
312+
Forms 3–5 map rows into structs but take a plain `&str` query — no parameter
313+
binding. [`query_params`](https://docs.rs/hyperdb-api/latest/hyperdb_api/struct.Connection.html#method.query_params)
314+
binds `$1`, `$2`, … placeholders safely but returns raw `Row`s. The `_as_params`
315+
methods are the intersection: they bind parameters via `ToSqlParam` *and* map
316+
each result row into a `FromRow` struct, in a single call — no manual
317+
`RowAccessor` loop and no SQL-injection risk.
318+
319+
There are three, mirroring the non-param trio:
320+
321+
- `fetch_one_as_params::<T>(query, params)` → `Result<T>`
322+
- `fetch_all_as_params::<T>(query, params)` → `Result<Vec<T>>`
323+
- `stream_as_params::<T>(query, params)` → `Result<impl Iterator<Item = Result<T>>>`
324+
(constant memory, like Form 5)
325+
326+
```rust
327+
use hyperdb_api::{Connection, CreateMode, FromRow, HyperProcess, Result};
328+
329+
fn main() -> Result<()> {
330+
let hyper = HyperProcess::new(None, None)?;
331+
let conn = Connection::new(&hyper, "products.hyper", CreateMode::DoNotCreate)?;
332+
333+
// Product derives FromRow (see Form 4); $1 is bound from `params`.
334+
let max_price = 15.0f64;
335+
let affordable: Vec<Product> = conn.fetch_all_as_params(
336+
"SELECT id, name, price, in_stock FROM products WHERE price < $1 ORDER BY id",
337+
&[&max_price],
338+
)?;
339+
for p in &affordable {
340+
println!("{:>2} {:<10} ${:.2} in_stock={}", p.id, p.name, p.price, p.in_stock);
341+
}
342+
Ok(())
343+
}
344+
```
345+
346+
Error handling matches the underlying methods: `fetch_one_as_params` /
347+
`fetch_all_as_params` return their errors on the `Result`; `stream_as_params`
348+
(sync) reports stream-open errors — including `FeatureNotSupported` on the gRPC
349+
transport, since prepared statements are TCP-only — on the outer `Result`, and
350+
per-row mapping errors as each item's `Result`. The async `stream_as_params`
351+
returns an `impl Stream` with no outer `Result`, so submission failures surface
352+
as the first yielded `Err` item (as with the async Form 5 above).
353+
354+
### Async
355+
356+
Each `_as_params` method has an `AsyncConnection` equivalent — `await` the
357+
`fetch_*` variants, and pin the `stream_as_params` stream just like Form 5.
358+
359+
---
360+
310361
## Choosing a form
311362

312363
| Need | Use |
@@ -316,6 +367,7 @@ async fn print_products(conn: &hyperdb_api::AsyncConnection) -> hyperdb_api::Res
316367
| Named struct, custom mapping logic | Form 3 (`impl FromRow` manually) |
317368
| Named struct, fields match columns | Form 4 (`#[derive(FromRow)]`) |
318369
| Streaming + named struct (constant memory) | Form 5 (`stream_as`) |
370+
| Named struct from a parameterized (`$1`) query | Form 6 (`fetch_*_as_params` / `stream_as_params`) |
319371

320372
For scalar values (a single `COUNT(*)`, `MAX`, etc.), use
321373
[`fetch_scalar`](https://docs.rs/hyperdb-api/latest/hyperdb_api/struct.Connection.html#method.fetch_scalar)

‎hyperdb-api/README.md‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -196,6 +196,22 @@ let rows_affected = conn.command_params(
196196
Supported parameter types: `i16`, `i32`, `i64`, `f32`, `f64`, `bool`, `&str`, `String`,
197197
`Option<T>`, `Date`, `Time`, `Timestamp`, `OffsetTimestamp`, `Vec<u8>`, `&[u8]`.
198198

199+
To map a parameterized query's results straight into a `FromRow` struct (instead
200+
of raw rows), use the `_as_params` variants — `fetch_one_as_params`,
201+
`fetch_all_as_params`, and `stream_as_params` (sync and async). They combine the
202+
`$1` binding above with the struct mapping of `fetch_*_as` / `stream_as` in one
203+
call:
204+
205+
```rust
206+
#[derive(hyperdb_api_derive::FromRow)]
207+
struct User { id: i32, name: String }
208+
209+
let users: Vec<User> = conn.fetch_all_as_params(
210+
"SELECT id, name FROM users WHERE org_id = $1",
211+
&[&42i32],
212+
)?;
213+
```
214+
199215
#### Compile-time SQL validation (opt-in)
200216

201217
With the `hyperdb-api-derive` crate's `compile-time` feature, the `query_as!`

‎hyperdb-api/examples/additional_examples/row_mapping_forms.rs‎

Lines changed: 42 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -13,10 +13,13 @@
1313
//! - **Form 3** — hand-written `FromRow` impl + `fetch_all_as`.
1414
//! - **Form 4** — `#[derive(FromRow)]` + `fetch_all_as`.
1515
//! - **Form 5** — streaming `FromRow`: `stream_as`, constant memory.
16+
//! - **Form 6** — parameterized `FromRow`: `fetch_all_as_params` /
17+
//! `stream_as_params`, combining `$1` parameter binding with struct mapping.
1618
//!
17-
//! Every form prints the same four products, so you can see they are
18-
//! equivalent. Form 5 is shown on both the sync `Connection` and the async
19+
//! Forms 1–5 print the same four products, so you can see they are equivalent.
20+
//! Form 5 is shown on both the sync `Connection` and the async
1921
//! `AsyncConnection` (which returns an `impl Stream` instead of an iterator).
22+
//! Form 6 binds a `$1` price filter, so it prints only the matching subset.
2023
//! Run with:
2124
//!
2225
//! cargo run -p hyperdb-api --example row_mapping_forms
@@ -51,6 +54,7 @@ fn main() -> Result<()> {
5154
form3_manual_from_row(&conn)?;
5255
form4_derive_from_row(&conn)?;
5356
form5_streaming_from_row(&conn)?;
57+
form6_parameterized_from_row(&conn)?;
5458

5559
// Form 5 also has an async flavor. Drop the sync connection first so the
5660
// async one reopens the same database file cleanly, then drive the stream
@@ -216,6 +220,42 @@ fn form5_streaming_from_row(conn: &Connection) -> Result<()> {
216220
Ok(())
217221
}
218222

223+
/// Form 6 — parameterized `FromRow`. The `_as_params` methods combine `$1`
224+
/// parameter binding (via `ToSqlParam`, exactly as `query_params`) with the
225+
/// automatic struct mapping of Forms 3–5. This closes the last gap: a
226+
/// parameterized `SELECT` whose rows you want as structs, in one call, with no
227+
/// manual `RowAccessor` loop and no SQL-injection risk.
228+
///
229+
/// `fetch_all_as_params` collects the matches; `stream_as_params` is the
230+
/// constant-memory streaming variant (shown here returning the same rows). Both
231+
/// have async equivalents on `AsyncConnection`.
232+
fn form6_parameterized_from_row(conn: &Connection) -> Result<()> {
233+
println!("== Form 6 — parameterized FromRow (fetch_all_as_params / stream_as_params) ==");
234+
235+
// Bind a price ceiling as $1 — only products cheaper than $15 come back.
236+
let max_price = 15.0f64;
237+
let affordable: Vec<ProductDerived> = conn.fetch_all_as_params(
238+
"SELECT id, name, price, in_stock FROM products WHERE price < $1 ORDER BY id",
239+
&[&max_price],
240+
)?;
241+
println!("fetch_all_as_params (price < ${max_price:.2}):");
242+
for p in &affordable {
243+
print_row(p.id, &p.name, p.price, p.in_stock);
244+
}
245+
246+
// Same query, streamed one chunk at a time via stream_as_params.
247+
println!("stream_as_params (same filter, constant memory):");
248+
for row_result in conn.stream_as_params::<ProductDerived>(
249+
"SELECT id, name, price, in_stock FROM products WHERE price < $1 ORDER BY id",
250+
&[&max_price],
251+
)? {
252+
let p = row_result?;
253+
print_row(p.id, &p.name, p.price, p.in_stock);
254+
}
255+
println!();
256+
Ok(())
257+
}
258+
219259
/// Form 5, async flavor. `AsyncConnection::stream_as` returns an
220260
/// `impl Stream<Item = Result<T>>` rather than an iterator — otherwise the
221261
/// shape is identical to the sync version: lazy, one chunk in memory at a

‎hyperdb-api/src/async_connection.rs‎

Lines changed: 147 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -465,6 +465,149 @@ impl AsyncConnection {
465465
}
466466
}
467467

468+
/// Fetches a single row from a **parameterized** query and maps it to a
469+
/// struct using [`FromRow`](crate::FromRow) (async).
470+
///
471+
/// Parameterized counterpart to [`fetch_one_as`](Self::fetch_one_as): binds
472+
/// `$1`, `$2`, … placeholders from `params` (via
473+
/// [`ToSqlParam`](crate::params::ToSqlParam), exactly as
474+
/// [`query_params`](Self::query_params)) and maps the first result row into
475+
/// `T`.
476+
///
477+
/// # Errors
478+
///
479+
/// - Returns [`Error::FeatureNotSupported`] on gRPC transports (prepared
480+
/// statements are TCP-only).
481+
/// - Returns the error from [`query_params`](Self::query_params) if the
482+
/// server rejects the statement, or on transport-level I/O failures.
483+
/// - Returns [`Error::Conversion`] with message `"Query returned no rows"`
484+
/// if the query produced zero rows.
485+
/// - Returns whatever [`FromRow::from_row`](crate::FromRow::from_row)
486+
/// produces when the row cannot be mapped.
487+
pub async fn fetch_one_as_params<T: crate::FromRow>(
488+
&self,
489+
query: &str,
490+
params: &[&dyn crate::params::ToSqlParam],
491+
) -> Result<T> {
492+
let row = self
493+
.query_params(query, params)
494+
.await?
495+
.require_first_row()
496+
.await?;
497+
let indices = row
498+
.schema()
499+
.map(crate::row_accessor::RowAccessor::build_indices)
500+
.unwrap_or_default();
501+
T::from_row(crate::RowAccessor::new(&row, &indices))
502+
}
503+
504+
/// Fetches all rows from a **parameterized** query and maps them to structs
505+
/// using [`FromRow`](crate::FromRow) (async).
506+
///
507+
/// Parameterized counterpart to [`fetch_all_as`](Self::fetch_all_as): binds
508+
/// `$1`, `$2`, … placeholders from `params` (see
509+
/// [`query_params`](Self::query_params)) and maps every result row into `T`.
510+
///
511+
/// # Errors
512+
///
513+
/// - Returns [`Error::FeatureNotSupported`] on gRPC transports.
514+
/// - Returns the error from [`query_params`](Self::query_params) if the
515+
/// server rejects the statement, or on transport-level I/O failures.
516+
/// - Returns the first error produced by
517+
/// [`FromRow::from_row`](crate::FromRow::from_row) on any row.
518+
pub async fn fetch_all_as_params<T: crate::FromRow>(
519+
&self,
520+
query: &str,
521+
params: &[&dyn crate::params::ToSqlParam],
522+
) -> Result<Vec<T>> {
523+
let rows = self
524+
.query_params(query, params)
525+
.await?
526+
.collect_rows()
527+
.await?;
528+
// Build the column-name → index lookup once from the first row's
529+
// schema; reuse for every row. See `fetch_all_as`.
530+
let indices = rows
531+
.first()
532+
.and_then(crate::result::Row::schema)
533+
.map(crate::row_accessor::RowAccessor::build_indices)
534+
.unwrap_or_default();
535+
rows.iter()
536+
.map(|r| T::from_row(crate::RowAccessor::new(r, &indices)))
537+
.collect()
538+
}
539+
540+
/// Returns a lazy `Stream` over the rows of a **parameterized** query,
541+
/// mapping each to `T` via [`FromRow`] (async).
542+
///
543+
/// Parameterized counterpart to [`stream_as`](Self::stream_as): binds `$1`,
544+
/// `$2`, … placeholders from `params` and streams the result with O(chunk)
545+
/// memory, mapping each row into `T`. The column-index map is built once on
546+
/// the first chunk and reused.
547+
///
548+
/// # Errors
549+
///
550+
/// Like [`stream_as`](Self::stream_as), this returns the `Stream` directly
551+
/// (no outer `Result`) — the query is lazy and does not execute until first
552+
/// polled. Each yielded item is a `Result<T>`:
553+
/// - The **first** item is `Err(e)` if statement submission fails:
554+
/// [`Error::FeatureNotSupported`] on gRPC transport, a `Parse`/`Bind`
555+
/// rejection, or a transport failure. These surface as the first item,
556+
/// *not* eagerly.
557+
/// - Subsequent items are `Ok(T)` on a clean per-row mapping, or `Err(e)`
558+
/// for a mapping failure (missing column, type mismatch, NULL in a
559+
/// non-`Option` field) or a transport error hit on a later chunk.
560+
///
561+
/// [`FromRow`]: crate::FromRow
562+
pub fn stream_as_params<'a, T: crate::FromRow + 'a>(
563+
&'a self,
564+
query: &str,
565+
params: &[&dyn crate::params::ToSqlParam],
566+
) -> impl futures_core::Stream<Item = Result<T>> + 'a {
567+
// `&[&dyn ToSqlParam]` can't cross the `try_stream!` await points, so
568+
// own the query string and encode params up front (encoding needs no
569+
// connection). The prepare+execute sequence below mirrors
570+
// `query_params` (see that method) — keep the two in sync if its
571+
// Parse/Bind/Execute handling ever changes.
572+
let query = query.to_owned();
573+
let oids: Vec<crate::Oid> = params.iter().map(|p| p.sql_oid()).collect();
574+
let encoded: Vec<Option<Vec<u8>>> = params.iter().map(|p| p.encode_param()).collect();
575+
async_stream::try_stream! {
576+
let client = match &self.transport {
577+
AsyncTransport::Tcp(tcp) => &tcp.client,
578+
AsyncTransport::Grpc(_) => {
579+
// `?` inside try_stream! yields Err(e) and terminates the
580+
// generator, so `unreachable!()` is dead code — its `!`
581+
// type just satisfies the arm's need for a `&AsyncClient`
582+
// (same type the Tcp arm produces).
583+
Err(Error::feature_not_supported(
584+
"prepared statements are not supported over gRPC transport",
585+
))?;
586+
unreachable!()
587+
}
588+
};
589+
let stmt = client.prepare_typed(&query, &oids).await?;
590+
let stream = client
591+
.execute_prepared_streaming(&stmt, encoded, crate::result::DEFAULT_BINARY_CHUNK_SIZE)
592+
.await?;
593+
let mut rs = AsyncRowset::from_prepared(stream).with_statement_guard(stmt);
594+
// The Prepared path captures the schema at prepare time, so the
595+
// column-name → index map is available immediately — build it once
596+
// up front rather than deferring to the first chunk (the empty-map
597+
// fallback matches `stream_as` / `fetch_all_as` if it is somehow
598+
// unavailable, surfacing a per-row `Missing` error).
599+
let idx = rs
600+
.schema()
601+
.map(|schema| crate::RowAccessor::build_owned_indices(&schema))
602+
.unwrap_or_default();
603+
while let Some(chunk) = rs.next_chunk().await? {
604+
for row in &chunk {
605+
yield T::from_row(crate::RowAccessor::new_owned(row, &idx))?;
606+
}
607+
}
608+
}
609+
}
610+
468611
/// Fetches a single non-NULL scalar value. Errors on empty / NULL.
469612
///
470613
/// # Errors
@@ -558,11 +701,12 @@ impl AsyncConnection {
558701
// Parameterized Queries
559702
// =========================================================================
560703

561-
/// Executes a parameterized query with safely escaped parameters (async).
704+
/// Executes a parameterized query with binary-encoded parameters (async).
562705
///
563706
/// Mirrors the sync [`Connection::query_params`](crate::Connection::query_params);
564-
/// see that method for the design rationale around text-mode escaping
565-
/// vs. future native Bind/Execute support.
707+
/// see that method for the design rationale. Parameters travel through the
708+
/// extended query protocol (Parse/Bind/Execute) in HyperBinary format — no
709+
/// SQL escaping, full SQL-injection safety regardless of parameter content.
566710
///
567711
/// # Errors
568712
///

0 commit comments

Comments
 (0)