hyperdb_api/result.rs
1// Copyright (c) 2026, Salesforce, Inc. All rights reserved.
2// SPDX-License-Identifier: Apache-2.0 OR MIT
3
4//! Query result handling with type-safe value access.
5//!
6//! This module provides types for working with query results:
7//! - [`Rowset`] — Streaming result set with memory-efficient chunked iteration
8//! - [`RowIterator`] — C++-like iterator for simple row-by-row processing
9//! - [`ResultSchema`] — Column metadata (names and types)
10//!
11//! # Streaming Design
12//!
13//! Query results are streamed from the server in chunks of up to
14//! [`DEFAULT_BINARY_CHUNK_SIZE`] rows (64K). Only one chunk is held in memory
15//! at a time, so memory usage is `O(chunk_size)` regardless of total result
16//! size — safe for billion-row results.
17//!
18//! # Iteration Patterns
19//!
20//! Two patterns are available, both streaming with constant memory:
21//!
22//! ## Pattern 1: Chunked (`next_chunk()`) — batch processing
23//!
24//! Best for high-throughput scenarios. Error checking happens once per chunk
25//! (~64K rows), and you get direct `Vec<Row>` iteration with good cache
26//! locality. Natural for batch operations, vectorized processing, or
27//! parallelizing across chunks.
28//!
29//! ```no_run
30//! # use hyperdb_api::{Connection, CreateMode, Result};
31//! # fn example(conn: &Connection) -> Result<()> {
32//! let mut result = conn.execute_query("SELECT * FROM table")?;
33//! while let Some(chunk) = result.next_chunk()? {
34//! for row in &chunk {
35//! let id: Option<i32> = row.get(0);
36//! let value: Option<f64> = row.get(1);
37//! }
38//! }
39//! # Ok(())
40//! # }
41//! ```
42//!
43//! ## Pattern 2: Iterator (`rows()`) — simple row-by-row
44//!
45//! Best for simple iteration where you process one row at a time. Each item
46//! is `Result<Row>` since chunk fetches can fail, so error checking happens
47//! per-row. The extra iterator wrapper adds slight overhead compared to
48//! `next_chunk()`.
49//!
50//! ```no_run
51//! # use hyperdb_api::{Connection, Result};
52//! # fn example(conn: &Connection) -> Result<()> {
53//! let result = conn.execute_query("SELECT * FROM table")?;
54//! for row in result.rows() {
55//! let row = row?; // Handle potential errors
56//! let id: Option<i32> = row.get(0);
57//! let value: Option<f64> = row.get(1);
58//! }
59//! # Ok(())
60//! # }
61//! ```
62//!
63//! **When to use which:**
64//! - `rows()` — simple iteration, one row at a time, small overhead acceptable
65//! - `next_chunk()` — maximum performance, large result sets, batch operations
66//!
67//! # Type Coercion
68//!
69//! The generic `row.get::<T>()` method supports automatic widening coercion:
70//!
71//! | Request Type | Coerces From |
72//! |---|---|
73//! | `i32` | `i16` |
74//! | `i64` | `i32`, `i16` |
75//! | `f64` | `f32` |
76//!
77//! Direct accessors (`row.get_i32()`, `row.get_f64()`) skip coercion for
78//! slightly better performance when the exact type is known.
79
80use std::sync::Arc;
81
82use arrow::array::Array;
83use arrow::record_batch::RecordBatch;
84use hyperdb_api_core::client::QueryStream;
85use hyperdb_api_core::client::StreamRow;
86use hyperdb_api_core::types::SqlType;
87
88use crate::arrow_result::{ArrowRowset, FromArrowValue};
89use crate::error::Result;
90
91/// Default chunk size for streaming queries (64K rows).
92pub(crate) const DEFAULT_BINARY_CHUNK_SIZE: usize = 65536;
93
94// =============================================================================
95// Row - Unified row type for both TCP and gRPC
96// =============================================================================
97
98/// A row from a query result, providing typed value access.
99///
100/// This type abstracts over the underlying transport (TCP or gRPC),
101/// providing a consistent API for accessing column values regardless
102/// of how the data was retrieved.
103///
104/// # Example
105///
106/// ```no_run
107/// # use hyperdb_api::Result;
108/// # fn example(result: hyperdb_api::Rowset) -> Result<()> {
109/// for row in result.rows() {
110/// let row = row?;
111/// let id: Option<i32> = row.get(0);
112/// let name: Option<String> = row.get(1);
113/// // Or use direct accessors
114/// let value = row.get_f64(2);
115/// }
116/// # Ok(())
117/// # }
118/// ```
119pub struct Row {
120 inner: RowInner,
121 /// Shared schema reference for the parent rowset. Every row
122 /// produced by [`Rowset::next_chunk`] carries this (cloned cheaply
123 /// from an `Arc`) so that metadata-dependent decoders like
124 /// [`Self::get_numeric`] can look up `SqlType` per column without
125 /// the caller plumbing scale through manually. `None` only in the
126 /// unusual case a row is constructed outside `next_chunk` (no such
127 /// path exists in-tree today; the field is `Option` so future
128 /// schemas-unavailable paths remain compilable).
129 schema: Option<Arc<ResultSchema>>,
130}
131
132impl std::fmt::Debug for Row {
133 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
134 f.debug_struct("Row")
135 .field("has_schema", &self.schema.is_some())
136 .finish_non_exhaustive()
137 }
138}
139
140/// Internal per-transport backing for a [`Row`]. Not public: all
141/// consumer-visible API goes through `Row`'s methods, which dispatch
142/// on this enum internally.
143enum RowInner {
144 /// Row from TCP transport (`StreamRow`).
145 Tcp(StreamRow),
146 /// Row from gRPC transport (Arrow-backed).
147 Arrow {
148 /// The record batch containing this row's data.
149 batch: Arc<RecordBatch>,
150 /// Index of this row within the batch.
151 row_index: usize,
152 },
153}
154
155impl Row {
156 /// Construct a TCP-backed row with an attached schema reference.
157 #[inline]
158 pub(crate) fn from_tcp(row: StreamRow, schema: Option<Arc<ResultSchema>>) -> Self {
159 Row {
160 inner: RowInner::Tcp(row),
161 schema,
162 }
163 }
164
165 /// Construct an Arrow-backed row with an attached schema reference.
166 #[inline]
167 pub(crate) fn from_arrow(
168 batch: Arc<RecordBatch>,
169 row_index: usize,
170 schema: Option<Arc<ResultSchema>>,
171 ) -> Self {
172 Row {
173 inner: RowInner::Arrow { batch, row_index },
174 schema,
175 }
176 }
177
178 /// Returns the schema this row belongs to, if attached.
179 ///
180 /// Every row produced by [`Rowset::next_chunk`] has a schema
181 /// attached — so this returns `Some` for any row obtained through
182 /// the public API.
183 #[inline]
184 pub fn schema(&self) -> Option<&ResultSchema> {
185 self.schema.as_deref()
186 }
187
188 /// Returns the `SqlType` of the column at the given index, if the
189 /// schema is attached and the index is in bounds.
190 ///
191 /// Useful for metadata-dependent decoders like [`Self::get_numeric`]
192 /// that need per-column precision and scale. Most callers reach for
193 /// [`Self::get`] / [`Self::try_get`] instead, which handle this
194 /// lookup internally via the [`RowValue`] trait.
195 #[inline]
196 pub fn sql_type(&self, idx: usize) -> Option<SqlType> {
197 let schema = self.schema.as_deref()?;
198 if idx < schema.column_count() {
199 Some(schema.column(idx).sql_type())
200 } else {
201 None
202 }
203 }
204
205 /// Gets a typed value at the given column index.
206 ///
207 /// # Example
208 ///
209 /// ```no_run
210 /// # use hyperdb_api::Row;
211 /// # fn example(row: &Row) {
212 /// let id: Option<i32> = row.get(0);
213 /// let name: Option<String> = row.get(1);
214 /// # }
215 /// ```
216 #[inline]
217 pub fn get<T: RowValue>(&self, idx: usize) -> Option<T> {
218 T::from_row(self, idx)
219 }
220
221 /// Gets a typed value at the given column index, returning a `Result`
222 /// with a descriptive error on failure.
223 ///
224 /// Use this in [`FromRow`] implementations for better error messages
225 /// than bare `row.get(idx).ok_or(...)`.
226 ///
227 /// # Example
228 ///
229 /// Most callers should reach for [`crate::FromRow`] +
230 /// [`crate::RowAccessor`] for typed mapping. `try_get` is the
231 /// underlying positional building block; useful when you need
232 /// indexed access from a hand-rolled loop.
233 ///
234 /// ```no_run
235 /// # use hyperdb_api::{Row, Result};
236 /// # fn read(row: &Row) -> Result<(i32, String)> {
237 /// let id: i32 = row.try_get(0, "id")?;
238 /// let name: String = row.try_get(1, "name")?;
239 /// # Ok((id, name))
240 /// # }
241 /// ```
242 ///
243 /// # Errors
244 ///
245 /// - Returns [`crate::Error::Conversion`] if `idx` is out of bounds for the row's
246 /// column count.
247 /// - Returns [`crate::Error::Conversion`] if the cell is SQL `NULL` or its value
248 /// cannot be decoded as `T`.
249 pub fn try_get<T: RowValue>(&self, idx: usize, column_name: &str) -> crate::error::Result<T> {
250 if idx >= self.column_count() {
251 return Err(crate::error::Error::conversion(format!(
252 "Column index {} ({:?}) out of bounds — row has {} columns",
253 idx,
254 column_name,
255 self.column_count(),
256 )));
257 }
258 self.get::<T>(idx).ok_or_else(|| {
259 crate::error::Error::conversion(format!(
260 "Column {idx} ({column_name:?}) is NULL or has incompatible type",
261 ))
262 })
263 }
264
265 /// Looks up a column by name and returns its value as `T`.
266 ///
267 /// Convenient for hand-coded paths that aren't using
268 /// [`FromRow`]. The lookup is a linear scan over
269 /// [`ResultSchema::column_index`]; for hot paths (many rows × many
270 /// fields), prefer
271 /// [`fetch_one_as`](crate::Connection::fetch_one_as) /
272 /// [`fetch_all_as`](crate::Connection::fetch_all_as), which build
273 /// a cached column-name → index lookup once per query and hand
274 /// every `FromRow` impl a [`RowAccessor`](crate::RowAccessor) that
275 /// reuses it.
276 ///
277 /// # Errors
278 ///
279 /// - [`crate::Error::Column`] with [`crate::ColumnErrorKind::Missing`]
280 /// if no column with `name` exists in the row's schema (or the
281 /// row has no schema attached).
282 /// - [`crate::Error::Conversion`] if the cell is `NULL` or cannot
283 /// be decoded as `T`. (Inherited from [`Self::try_get`].)
284 pub fn get_by_name<T: RowValue>(&self, name: &str) -> crate::error::Result<T> {
285 let idx = self
286 .schema()
287 .and_then(|s| s.column_index(name))
288 .ok_or_else(|| {
289 crate::error::Error::column(name, crate::error::ColumnErrorKind::Missing)
290 })?;
291 self.try_get(idx, name)
292 }
293
294 /// Returns an Arrow column reference, or `None` if the index is out of bounds.
295 ///
296 /// This is a safe wrapper around `batch.column(idx)` that avoids panicking.
297 #[inline]
298 fn arrow_column(batch: &RecordBatch, idx: usize) -> Option<&Arc<dyn Array>> {
299 if idx < batch.num_columns() {
300 Some(batch.column(idx))
301 } else {
302 None
303 }
304 }
305
306 /// Gets an i16 value at the given column index.
307 #[inline]
308 pub fn get_i16(&self, idx: usize) -> Option<i16> {
309 match &self.inner {
310 RowInner::Tcp(row) => row.get_i16(idx),
311 RowInner::Arrow { batch, row_index } => {
312 i16::from_arrow_column(Self::arrow_column(batch, idx)?, *row_index)
313 }
314 }
315 }
316
317 /// Gets an i32 value at the given column index.
318 #[inline]
319 pub fn get_i32(&self, idx: usize) -> Option<i32> {
320 match &self.inner {
321 RowInner::Tcp(row) => row.get_i32(idx),
322 RowInner::Arrow { batch, row_index } => {
323 i32::from_arrow_column(Self::arrow_column(batch, idx)?, *row_index)
324 }
325 }
326 }
327
328 /// Gets an i64 value at the given column index.
329 #[inline]
330 pub fn get_i64(&self, idx: usize) -> Option<i64> {
331 match &self.inner {
332 RowInner::Tcp(row) => row.get_i64(idx),
333 RowInner::Arrow { batch, row_index } => {
334 i64::from_arrow_column(Self::arrow_column(batch, idx)?, *row_index)
335 }
336 }
337 }
338
339 /// Gets an f32 value at the given column index.
340 #[inline]
341 pub fn get_f32(&self, idx: usize) -> Option<f32> {
342 match &self.inner {
343 RowInner::Tcp(row) => row.get_f32(idx),
344 RowInner::Arrow { batch, row_index } => {
345 f32::from_arrow_column(Self::arrow_column(batch, idx)?, *row_index)
346 }
347 }
348 }
349
350 /// Gets an f64 value at the given column index.
351 #[inline]
352 pub fn get_f64(&self, idx: usize) -> Option<f64> {
353 match &self.inner {
354 RowInner::Tcp(row) => row.get_f64(idx),
355 RowInner::Arrow { batch, row_index } => {
356 f64::from_arrow_column(Self::arrow_column(batch, idx)?, *row_index)
357 }
358 }
359 }
360
361 /// Gets a bool value at the given column index.
362 #[inline]
363 pub fn get_bool(&self, idx: usize) -> Option<bool> {
364 match &self.inner {
365 RowInner::Tcp(row) => row.get_bool(idx),
366 RowInner::Arrow { batch, row_index } => {
367 bool::from_arrow_column(Self::arrow_column(batch, idx)?, *row_index)
368 }
369 }
370 }
371
372 /// Gets a String value at the given column index.
373 #[inline]
374 pub fn get_string(&self, idx: usize) -> Option<String> {
375 match &self.inner {
376 RowInner::Tcp(row) => row.get_string(idx),
377 RowInner::Arrow { batch, row_index } => {
378 String::from_arrow_column(Self::arrow_column(batch, idx)?, *row_index)
379 }
380 }
381 }
382
383 /// Checks if the value at the given column is null.
384 #[inline]
385 pub fn is_null(&self, idx: usize) -> bool {
386 match &self.inner {
387 RowInner::Tcp(row) => row.is_null(idx),
388 RowInner::Arrow { batch, row_index } => match Self::arrow_column(batch, idx) {
389 Some(col) => col.is_null(*row_index),
390 None => true,
391 },
392 }
393 }
394
395 /// Returns the number of columns in this row.
396 #[inline]
397 pub fn column_count(&self) -> usize {
398 match &self.inner {
399 RowInner::Tcp(row) => row.column_count(),
400 RowInner::Arrow { batch, .. } => batch.num_columns(),
401 }
402 }
403
404 /// Gets raw bytes at the given column index.
405 ///
406 /// For TCP rows, returns the raw binary data. For Arrow rows, this method
407 /// is not available and returns None.
408 #[inline]
409 pub fn get_bytes(&self, idx: usize) -> Option<Vec<u8>> {
410 match &self.inner {
411 RowInner::Tcp(row) => row.get_bytes(idx).map(<[u8]>::to_vec),
412 RowInner::Arrow { batch, row_index } => {
413 Vec::<u8>::from_arrow_column(Self::arrow_column(batch, idx)?, *row_index)
414 }
415 }
416 }
417
418 /// Gets a Date value at the given column index.
419 #[inline]
420 pub fn get_date(&self, idx: usize) -> Option<hyperdb_api_core::types::Date> {
421 match &self.inner {
422 RowInner::Tcp(row) => row.get(idx),
423 RowInner::Arrow { batch, row_index } => {
424 // Arrow Date32 is days since Unix epoch (1970-01-01)
425 // Hyper Date is days since Hyper epoch (2000-01-01)
426 use arrow::array::Date32Array;
427 let col = Self::arrow_column(batch, idx)?;
428 let arr = col.as_any().downcast_ref::<Date32Array>()?;
429 if arr.is_null(*row_index) {
430 return None;
431 }
432 let unix_days = arr.value(*row_index);
433 // Convert from Unix epoch to Hyper epoch (diff is 10957 days)
434 let hyper_days = unix_days - 10957;
435 Some(hyperdb_api_core::types::Date::from_days(hyper_days))
436 }
437 }
438 }
439
440 /// Gets a Time value at the given column index.
441 #[inline]
442 pub fn get_time(&self, idx: usize) -> Option<hyperdb_api_core::types::Time> {
443 match &self.inner {
444 RowInner::Tcp(row) => row.get(idx),
445 RowInner::Arrow { batch, row_index } => {
446 // Arrow Time64 is microseconds since midnight
447 use arrow::array::Time64MicrosecondArray;
448 let col = Self::arrow_column(batch, idx)?;
449 let arr = col.as_any().downcast_ref::<Time64MicrosecondArray>()?;
450 if arr.is_null(*row_index) {
451 return None;
452 }
453 let micros = u64::try_from(arr.value(*row_index)).ok()?;
454 Some(hyperdb_api_core::types::Time::from_microseconds(micros))
455 }
456 }
457 }
458
459 /// Gets a Timestamp value at the given column index.
460 #[inline]
461 pub fn get_timestamp(&self, idx: usize) -> Option<hyperdb_api_core::types::Timestamp> {
462 match &self.inner {
463 RowInner::Tcp(row) => row.get(idx),
464 RowInner::Arrow { batch, row_index } => {
465 // Arrow Timestamp is microseconds since Unix epoch
466 // Hyper Timestamp is microseconds since Hyper epoch (2000-01-01)
467 use arrow::array::TimestampMicrosecondArray;
468 let col = Self::arrow_column(batch, idx)?;
469 let arr = col.as_any().downcast_ref::<TimestampMicrosecondArray>()?;
470 if arr.is_null(*row_index) {
471 return None;
472 }
473 let unix_micros = arr.value(*row_index);
474 // Convert from Unix epoch to Hyper epoch
475 // 2000-01-01 is 946684800 seconds after 1970-01-01
476 let hyper_micros = unix_micros - 946_684_800_000_000;
477 Some(hyperdb_api_core::types::Timestamp::from_microseconds(
478 hyper_micros,
479 ))
480 }
481 }
482 }
483
484 /// Gets an `OffsetTimestamp` (TIMESTAMP WITH TIME ZONE) value at the given column index.
485 #[inline]
486 pub fn get_offset_timestamp(
487 &self,
488 idx: usize,
489 ) -> Option<hyperdb_api_core::types::OffsetTimestamp> {
490 match &self.inner {
491 RowInner::Tcp(row) => row.get(idx),
492 RowInner::Arrow { batch, row_index } => {
493 // Arrow TimestampTz is microseconds since Unix epoch with timezone
494 use arrow::array::TimestampMicrosecondArray;
495 let col = Self::arrow_column(batch, idx)?;
496 let arr = col.as_any().downcast_ref::<TimestampMicrosecondArray>()?;
497 if arr.is_null(*row_index) {
498 return None;
499 }
500 let unix_micros = arr.value(*row_index);
501 let hyper_micros = unix_micros - 946_684_800_000_000;
502 let ts = hyperdb_api_core::types::Timestamp::from_microseconds(hyper_micros);
503 Some(hyperdb_api_core::types::OffsetTimestamp::new(ts, 0))
504 }
505 }
506 }
507
508 /// Gets an Interval value at the given column index.
509 #[inline]
510 pub fn get_interval(&self, idx: usize) -> Option<hyperdb_api_core::types::Interval> {
511 match &self.inner {
512 RowInner::Tcp(row) => row.get(idx),
513 RowInner::Arrow { batch, row_index } => {
514 // Arrow MonthDayNano interval → Hyper Interval
515 use arrow::array::IntervalMonthDayNanoArray;
516 let col = Self::arrow_column(batch, idx)?;
517 let arr = col.as_any().downcast_ref::<IntervalMonthDayNanoArray>()?;
518 if arr.is_null(*row_index) {
519 return None;
520 }
521 let v = arr.value(*row_index);
522 let micros = v.nanoseconds / 1000;
523 Some(hyperdb_api_core::types::Interval::new(
524 v.months, v.days, micros,
525 ))
526 }
527 }
528 }
529
530 /// Gets a `NUMERIC` value at the given column index.
531 ///
532 /// This is the metadata-aware variant of [`Self::get_bytes`] +
533 /// [`hyperdb_api_core::types::Numeric::from_binary_with_scale`]: it looks up
534 /// the column's `SqlType::Numeric { scale, .. }` from the attached
535 /// schema and decodes the wire bytes with that scale, handling
536 /// both of Hyper's NUMERIC wire forms transparently:
537 ///
538 /// - **8 bytes** (i64) when the column's declared precision ≤ 18
539 /// (Hyper's `Type::Numeric`). This is what aggregates like
540 /// `AVG(INTEGER)` return as `Numeric(16, 6)`.
541 /// - **16 bytes** (i128) when declared precision > 18
542 /// (Hyper's `Type::BigNumeric`).
543 ///
544 /// Returns `None` if any of the following are true: the value is
545 /// NULL, the schema isn't attached (which never happens for rows
546 /// obtained through [`Rowset::next_chunk`]), the column at `idx`
547 /// isn't `NUMERIC`, or the bytes can't be decoded.
548 ///
549 /// For non-TCP (Arrow/gRPC) rows, this path falls back to reading
550 /// the Arrow-native `Decimal128` / `Decimal256` columns; the scale
551 /// lives in the Arrow type descriptor in that case.
552 pub fn get_numeric(&self, idx: usize) -> Option<hyperdb_api_core::types::Numeric> {
553 match &self.inner {
554 RowInner::Tcp(_) => {
555 // TCP: decode raw bytes with scale from the schema.
556 //
557 // `SqlType::Numeric::scale` is `u32` and Hyper's own
558 // `NUMERIC(p, s)` caps at `p ≤ 38` (per
559 // `hyper/rts/type/Type.hpp`), so any legitimate scale
560 // fits easily in `u8`. But `scale as u8` silently
561 // truncates the high bits for values > 255, and a
562 // malformed server response or a bug in typemod
563 // parsing could deliver such a value — at which point
564 // we'd produce a `Numeric` with the wrong (truncated)
565 // scale and no error signal. `u8::try_from` returns
566 // `Err` for out-of-range, `?` propagates `None`, and
567 // the caller gets a clean "no value" instead of
568 // silent corruption. Symmetric with the Arrow
569 // negative-scale guard a few lines below.
570 let scale: u8 = match self.sql_type(idx)? {
571 SqlType::Numeric { scale, .. } => u8::try_from(scale).ok()?,
572 _ => return None,
573 };
574 let bytes = self.get_bytes(idx)?;
575 hyperdb_api_core::types::Numeric::from_binary_with_scale(&bytes, scale).ok()
576 }
577 RowInner::Arrow { batch, row_index } => {
578 use arrow::array::{Decimal128Array, Decimal256Array};
579 use arrow::datatypes::DataType as ArrowType;
580 let col = Self::arrow_column(batch, idx)?;
581 // Arrow stores decimal precision/scale in the type
582 // descriptor itself, so there's no separate schema
583 // lookup needed on this path.
584 //
585 // Note: Arrow's decimal scale is `i8` and can legally
586 // be negative (negative scale = "value is multiplied
587 // by 10^abs(scale)", e.g. scale=-2 on raw=5 renders
588 // as 500). Hyper's `Numeric` uses `u8` scale and has
589 // no representation for the negative-scale
590 // multiplier. Rather than silently dropping the
591 // multiplier (which would make raw=5 display as 5
592 // instead of 500), we surface it as "no value" via
593 // `try_into` + `?`. Negative-scale decimals don't
594 // originate from Hyper's own gRPC encoder — but
595 // `Row` can be fed from externally-loaded Arrow
596 // files, so defensive handling costs nothing and
597 // prevents a silent-corruption failure mode.
598 match col.data_type() {
599 ArrowType::Decimal128(_precision, scale) => {
600 let scale_u8: u8 = (*scale).try_into().ok()?;
601 let arr = col.as_any().downcast_ref::<Decimal128Array>()?;
602 if arr.is_null(*row_index) {
603 return None;
604 }
605 let raw = arr.value(*row_index); // i128
606 Some(hyperdb_api_core::types::Numeric::new(raw, scale_u8))
607 }
608 ArrowType::Decimal256(_precision, scale) => {
609 // i256 from Arrow; Hyper NUMERIC caps at i128
610 // (precision ≤ 38). Narrow to i128; this is
611 // lossless for any value Hyper would actually
612 // produce. Values outside that range are a
613 // server-side contract violation.
614 let scale_u8: u8 = (*scale).try_into().ok()?;
615 let arr = col.as_any().downcast_ref::<Decimal256Array>()?;
616 if arr.is_null(*row_index) {
617 return None;
618 }
619 let raw = arr.value(*row_index);
620 let as_i128: i128 = raw.to_i128()?;
621 Some(hyperdb_api_core::types::Numeric::new(as_i128, scale_u8))
622 }
623 _ => None,
624 }
625 }
626 }
627 }
628}
629
630/// Trait for types that can be extracted from a Row.
631pub trait RowValue: Sized {
632 /// Extract a value from a Row at the given column index.
633 fn from_row(row: &Row, idx: usize) -> Option<Self>;
634}
635
636impl RowValue for i16 {
637 #[inline]
638 fn from_row(row: &Row, idx: usize) -> Option<Self> {
639 row.get_i16(idx)
640 }
641}
642
643impl RowValue for i32 {
644 #[inline]
645 fn from_row(row: &Row, idx: usize) -> Option<Self> {
646 row.get_i32(idx).or_else(|| row.get_i16(idx).map(i32::from))
647 }
648}
649
650impl RowValue for i64 {
651 #[inline]
652 fn from_row(row: &Row, idx: usize) -> Option<Self> {
653 row.get_i64(idx)
654 .or_else(|| row.get_i32(idx).map(i64::from))
655 .or_else(|| row.get_i16(idx).map(i64::from))
656 }
657}
658
659impl RowValue for f32 {
660 #[inline]
661 fn from_row(row: &Row, idx: usize) -> Option<Self> {
662 row.get_f32(idx)
663 }
664}
665
666impl RowValue for f64 {
667 #[inline]
668 fn from_row(row: &Row, idx: usize) -> Option<Self> {
669 row.get_f64(idx).or_else(|| row.get_f32(idx).map(f64::from))
670 }
671}
672
673impl RowValue for bool {
674 #[inline]
675 fn from_row(row: &Row, idx: usize) -> Option<Self> {
676 row.get_bool(idx)
677 }
678}
679
680impl RowValue for String {
681 #[inline]
682 fn from_row(row: &Row, idx: usize) -> Option<Self> {
683 row.get_string(idx)
684 }
685}
686
687impl RowValue for Vec<u8> {
688 #[inline]
689 fn from_row(row: &Row, idx: usize) -> Option<Self> {
690 row.get_bytes(idx)
691 }
692}
693
694impl RowValue for hyperdb_api_core::types::Date {
695 #[inline]
696 fn from_row(row: &Row, idx: usize) -> Option<Self> {
697 row.get_date(idx)
698 }
699}
700
701impl RowValue for hyperdb_api_core::types::Time {
702 #[inline]
703 fn from_row(row: &Row, idx: usize) -> Option<Self> {
704 row.get_time(idx)
705 }
706}
707
708impl RowValue for hyperdb_api_core::types::Timestamp {
709 #[inline]
710 fn from_row(row: &Row, idx: usize) -> Option<Self> {
711 row.get_timestamp(idx)
712 }
713}
714
715impl RowValue for hyperdb_api_core::types::OffsetTimestamp {
716 #[inline]
717 fn from_row(row: &Row, idx: usize) -> Option<Self> {
718 row.get_offset_timestamp(idx)
719 }
720}
721
722impl RowValue for hyperdb_api_core::types::Interval {
723 #[inline]
724 fn from_row(row: &Row, idx: usize) -> Option<Self> {
725 row.get_interval(idx)
726 }
727}
728
729impl RowValue for hyperdb_api_core::types::Numeric {
730 /// Unlike every other `RowValue` impl, `Numeric` decode requires
731 /// per-column metadata (scale + wire-form width) that lives on the
732 /// row's attached `ResultSchema`. [`Row::get_numeric`] does the
733 /// lookup; this impl delegates there so generic `row.get::<Numeric>()`
734 /// / `row.try_get::<Numeric>(idx, "name")` call sites work the same
735 /// as every other type.
736 #[inline]
737 fn from_row(row: &Row, idx: usize) -> Option<Self> {
738 row.get_numeric(idx)
739 }
740}
741
742// =============================================================================
743// FromRow - Struct mapping trait
744// =============================================================================
745
746/// Trait for types that can be constructed from a database row.
747///
748/// Used by [`Connection::fetch_one_as`](crate::Connection::fetch_one_as)
749/// and [`Connection::fetch_all_as`](crate::Connection::fetch_all_as)
750/// to map query results into typed structs. Implementations receive
751/// a [`RowAccessor`](crate::RowAccessor), which provides name-based
752/// access via a column-name → index lookup built once per query.
753///
754/// # Recommended: derive
755///
756/// In most cases the `#[derive(FromRow)]` macro handles the mapping
757/// for you — match struct field names to column names automatically,
758/// with `#[hyperdb(rename = "...")]` for cases where they differ:
759///
760/// ```ignore
761/// use hyperdb_api::FromRow;
762///
763/// #[derive(FromRow)]
764/// struct User {
765/// id: i32,
766/// name: String,
767/// #[hyperdb(rename = "email_address")]
768/// email: Option<String>,
769/// }
770/// ```
771///
772/// # Hand-written impl
773///
774/// For custom mapping logic (computed fields, multi-column composition,
775/// etc.) implement the trait directly:
776///
777/// ```no_run
778/// use hyperdb_api::{FromRow, RowAccessor, Result};
779///
780/// struct User { id: i32, name: String, active: bool }
781///
782/// impl FromRow for User {
783/// fn from_row(row: RowAccessor<'_>) -> Result<Self> {
784/// Ok(User {
785/// id: row.get("id")?,
786/// name: row.get("name")?,
787/// active: row.get("active")?,
788/// })
789/// }
790/// }
791/// ```
792///
793/// For ad-hoc tuple destructuring of small results, use
794/// [`Row::get`](crate::Row::get) directly — there are no blanket
795/// tuple `FromRow` impls. Define a struct with `#[derive(FromRow)]`
796/// for typed access in `fetch_*_as`.
797pub trait FromRow: Sized {
798 /// Constructs an instance from a database row.
799 ///
800 /// # Errors
801 ///
802 /// Returns an [`Error`](crate::Error) — typically
803 /// [`crate::Error::Column`] — when a required column is missing,
804 /// SQL `NULL`, or cannot be decoded as the expected type.
805 /// Implementations decide the exact failure shape.
806 fn from_row(row: crate::RowAccessor<'_>) -> crate::error::Result<Self>;
807}
808
809// =============================================================================
810// ResultSchema and ResultColumn
811// =============================================================================
812
813/// Metadata about a column in a result schema.
814#[derive(Debug, Clone)]
815pub struct ResultColumn {
816 /// The column name.
817 name: String,
818 /// The SQL type of the column.
819 sql_type: SqlType,
820 /// The column index (0-based).
821 index: usize,
822}
823
824impl ResultColumn {
825 /// Creates a new result column.
826 pub fn new(name: impl Into<String>, sql_type: SqlType, index: usize) -> Self {
827 ResultColumn {
828 name: name.into(),
829 sql_type,
830 index,
831 }
832 }
833
834 /// Returns the column name.
835 #[must_use]
836 pub fn name(&self) -> &str {
837 &self.name
838 }
839
840 /// Returns the SQL type of the column.
841 #[must_use]
842 pub fn sql_type(&self) -> SqlType {
843 self.sql_type
844 }
845
846 /// Returns the column index (0-based).
847 #[must_use]
848 pub fn index(&self) -> usize {
849 self.index
850 }
851}
852
853/// Schema information for a query result.
854///
855/// Provides metadata about the columns returned by a query, including
856/// column names and types.
857#[derive(Debug, Clone, Default)]
858pub struct ResultSchema {
859 columns: Vec<ResultColumn>,
860}
861
862impl ResultSchema {
863 /// Creates a new empty result schema.
864 #[must_use]
865 pub fn new() -> Self {
866 ResultSchema {
867 columns: Vec::new(),
868 }
869 }
870
871 /// Creates a result schema from column definitions.
872 #[must_use]
873 pub fn from_columns(columns: Vec<ResultColumn>) -> Self {
874 ResultSchema { columns }
875 }
876
877 /// Adds a column to the schema.
878 pub fn add_column(&mut self, name: impl Into<String>, sql_type: SqlType) {
879 let index = self.columns.len();
880 self.columns.push(ResultColumn::new(name, sql_type, index));
881 }
882
883 /// Returns the number of columns.
884 #[must_use]
885 pub fn column_count(&self) -> usize {
886 self.columns.len()
887 }
888
889 /// Returns all columns.
890 #[must_use]
891 pub fn columns(&self) -> &[ResultColumn] {
892 &self.columns
893 }
894
895 /// Returns the column at the given index.
896 ///
897 /// # Panics
898 ///
899 /// Panics if the index is out of bounds.
900 #[must_use]
901 pub fn column(&self, index: usize) -> &ResultColumn {
902 &self.columns[index]
903 }
904
905 /// Returns the column with the given name, if it exists.
906 #[must_use]
907 pub fn column_by_name(&self, name: &str) -> Option<&ResultColumn> {
908 self.columns.iter().find(|c| c.name == name)
909 }
910
911 /// Returns the index of the column with the given name, if it exists.
912 #[must_use]
913 pub fn column_index(&self, name: &str) -> Option<usize> {
914 self.columns.iter().position(|c| c.name == name)
915 }
916}
917
918// =============================================================================
919// Rowset (Streaming)
920// =============================================================================
921
922/// A streaming result set from a SQL query.
923///
924/// `Rowset` provides memory-efficient streaming access to query results.
925/// Results are fetched on-demand in chunks, keeping memory usage constant
926/// regardless of result set size. This makes it safe for any result size,
927/// from a single row to billions of rows.
928///
929/// # Example
930///
931/// ```no_run
932/// # use hyperdb_api::{Connection, Result};
933/// # fn example(conn: &Connection) -> Result<()> {
934/// let mut result = conn.execute_query("SELECT * FROM big_table")?;
935/// while let Some(chunk) = result.next_chunk()? {
936/// for row in &chunk {
937/// // Generic typed access (like C++ row.get<T>())
938/// let id: Option<i32> = row.get(0);
939/// let value: Option<f64> = row.get(1);
940///
941/// // Or direct accessors for performance
942/// let id = row.get_i32(0);
943/// let value = row.get_f64(1);
944/// }
945/// }
946/// # Ok(())
947/// # }
948/// ```
949///
950/// # Memory Behavior
951///
952/// - Only one chunk is held in memory at a time
953/// - Default chunk size is 64K rows (~few MB depending on row width)
954/// - Memory usage is `O(chunk_size)`, not `O(total_rows)`
955/// - Safe for billion-row results
956pub struct Rowset<'conn> {
957 inner: RowsetInner<'conn>,
958 /// Cached schema for this rowset, built lazily the first time
959 /// [`Self::next_chunk`] produces a non-empty chunk (TCP path — at
960 /// which point the `RowDescription` message has been observed) or
961 /// on first Arrow chunk (gRPC path). Stored as `Arc` so each row
962 /// produced by `next_chunk` gets a cheap ref-count clone — that's
963 /// how metadata-dependent decoders like [`Row::get_numeric`] reach
964 /// the column's `SqlType` without the caller plumbing scale
965 /// through manually.
966 schema_cache: Option<Arc<ResultSchema>>,
967 /// For one-shot prepared statements (the internal
968 /// [`crate::Connection::query_params`] path), hold the statement
969 /// handle here so its `Drop`-time `close_statement` fires *after*
970 /// the rowset releases its connection lock. Dropping the statement
971 /// before the rowset would deadlock because the inner stream owns
972 /// the connection's `MutexGuard`.
973 _statement_guard: Option<hyperdb_api_core::client::OwnedPreparedStatement>,
974}
975
976impl std::fmt::Debug for Rowset<'_> {
977 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
978 f.debug_struct("Rowset")
979 .field("has_schema_cache", &self.schema_cache.is_some())
980 .finish_non_exhaustive()
981 }
982}
983
984/// Internal enum to hold either TCP stream or Arrow data.
985enum RowsetInner<'conn> {
986 /// TCP streaming result (uses `QueryStream`).
987 Tcp(QueryStream<'conn>),
988 /// Arrow-based result from gRPC (all data loaded).
989 Arrow(ArrowRowset),
990 /// TCP streaming result from a prepared-statement execute.
991 Prepared(hyperdb_api_core::client::PreparedQueryStream<'conn>),
992}
993
994impl<'conn> Rowset<'conn> {
995 /// Creates a new Rowset from a `QueryStream` (TCP).
996 pub(crate) fn new(stream: QueryStream<'conn>) -> Self {
997 Rowset {
998 inner: RowsetInner::Tcp(stream),
999 schema_cache: None,
1000 _statement_guard: None,
1001 }
1002 }
1003
1004 /// Creates a new Rowset from Arrow IPC data (gRPC).
1005 pub(crate) fn from_arrow(arrow_rowset: ArrowRowset) -> Self {
1006 Rowset {
1007 inner: RowsetInner::Arrow(arrow_rowset),
1008 schema_cache: None,
1009 _statement_guard: None,
1010 }
1011 }
1012
1013 /// Creates a new Rowset from a prepared-statement streaming result.
1014 pub(crate) fn from_prepared(
1015 stream: hyperdb_api_core::client::PreparedQueryStream<'conn>,
1016 ) -> Self {
1017 Rowset {
1018 inner: RowsetInner::Prepared(stream),
1019 schema_cache: None,
1020 _statement_guard: None,
1021 }
1022 }
1023
1024 #[expect(
1025 clippy::used_underscore_binding,
1026 reason = "underscore-prefixed parameter retained for trait-method signature compatibility"
1027 )]
1028 /// Attaches a `OwnedPreparedStatement` that should be dropped
1029 /// **after** this rowset is consumed. Used by the one-shot
1030 /// prepare+execute path inside
1031 /// [`crate::Connection::query_params`] so the statement's
1032 /// Drop-time close doesn't deadlock on the rowset's still-held
1033 /// connection lock.
1034 pub(crate) fn with_statement_guard(
1035 mut self,
1036 statement: hyperdb_api_core::client::OwnedPreparedStatement,
1037 ) -> Self {
1038 self._statement_guard = Some(statement);
1039 self
1040 }
1041
1042 /// Returns the schema (column metadata) for the result set.
1043 ///
1044 /// For TCP connections, the schema is captured from the `RowDescription` message
1045 /// after the first chunk is read. For gRPC connections, the schema is available
1046 /// immediately from the Arrow data.
1047 ///
1048 /// Returns `None` if no data has been read yet (TCP only).
1049 ///
1050 /// # Example
1051 ///
1052 /// ```no_run
1053 /// # use hyperdb_api::{Connection, Result};
1054 /// # fn example(conn: &Connection) -> Result<()> {
1055 /// let mut result = conn.execute_query("SELECT id, name FROM users")?;
1056 /// // Read first chunk to capture schema (TCP) or get it immediately (gRPC)
1057 /// let _ = result.next_chunk()?;
1058 /// if let Some(schema) = result.schema() {
1059 /// for col in schema.columns() {
1060 /// println!("Column: {} ({})", col.name(), col.sql_type());
1061 /// }
1062 /// }
1063 /// # Ok(())
1064 /// # }
1065 /// ```
1066 #[must_use]
1067 pub fn schema(&self) -> Option<ResultSchema> {
1068 // Fast path: cache already populated by a previous call or by
1069 // `next_chunk`. Clone from the Arc so external callers get an
1070 // owned value independent of internal lifetimes.
1071 if let Some(ref cached) = self.schema_cache {
1072 return Some((**cached).clone());
1073 }
1074 // Slow path: schema hasn't been materialized yet. Build it from
1075 // the transport without populating the cache — `schema()` takes
1076 // `&self`, so mutation isn't possible here. `next_chunk` does
1077 // the caching pass for the row-construction hot path; if a
1078 // caller really only wants the schema and never touches rows,
1079 // they pay one build per call but this is rarely the pattern.
1080 self.build_schema()
1081 }
1082
1083 /// Compute the current schema without populating the cache.
1084 ///
1085 /// Pulls column metadata from the underlying transport and
1086 /// constructs a fresh `ResultSchema`. TCP builds `SqlType` via
1087 /// [`SqlType::from_oid_and_modifier`] so
1088 /// `NUMERIC(precision, scale)` and `VARCHAR(n)` recover their
1089 /// declared parameters from the `RowDescription` `atttypmod`
1090 /// field — dropping the modifier (which bare
1091 /// [`SqlType::from_oid`] does) silently turns every `NUMERIC`
1092 /// into `(precision: 0, scale: 0)` and corrupts decimal decodes
1093 /// downstream. Arrow comes pre-typed via
1094 /// `arrow_type_to_sql_type`.
1095 fn build_schema(&self) -> Option<ResultSchema> {
1096 match &self.inner {
1097 RowsetInner::Tcp(stream) => stream.schema().map(|cols| {
1098 let columns = cols
1099 .iter()
1100 .enumerate()
1101 .map(|(idx, col)| {
1102 let sql_type =
1103 SqlType::from_oid_and_modifier(col.type_oid().0, col.type_modifier());
1104 ResultColumn::new(col.name(), sql_type, idx)
1105 })
1106 .collect();
1107 ResultSchema::from_columns(columns)
1108 }),
1109 RowsetInner::Arrow(arrow) => {
1110 let schema = arrow.schema();
1111 let columns = schema
1112 .fields()
1113 .iter()
1114 .enumerate()
1115 .map(|(idx, field)| {
1116 ResultColumn::new(
1117 field.name(),
1118 crate::arrow_result::arrow_type_to_sql_type(field.data_type()),
1119 idx,
1120 )
1121 })
1122 .collect();
1123 Some(ResultSchema::from_columns(columns))
1124 }
1125 // Prepared statements: schema was captured at prepare time,
1126 // so it is always available immediately.
1127 RowsetInner::Prepared(stream) => {
1128 let cols = stream.schema();
1129 let columns = cols
1130 .iter()
1131 .enumerate()
1132 .map(|(idx, col)| {
1133 let sql_type =
1134 SqlType::from_oid_and_modifier(col.type_oid().0, col.type_modifier());
1135 ResultColumn::new(col.name(), sql_type, idx)
1136 })
1137 .collect();
1138 Some(ResultSchema::from_columns(columns))
1139 }
1140 }
1141 }
1142
1143 /// Populate `schema_cache` if not yet set, then return an `Arc`
1144 /// clone of the cached schema for row construction. Called by
1145 /// `next_chunk` so every row produced gets a cheap schema
1146 /// reference without re-building the `ResultSchema` per chunk.
1147 fn cached_schema_arc(&mut self) -> Option<Arc<ResultSchema>> {
1148 if self.schema_cache.is_none() {
1149 if let Some(schema) = self.build_schema() {
1150 self.schema_cache = Some(Arc::new(schema));
1151 }
1152 }
1153 self.schema_cache.clone()
1154 }
1155
1156 /// Returns the next chunk of rows from the result set.
1157 ///
1158 /// Each chunk contains up to `chunk_size` rows (default 64K).
1159 /// Returns `Ok(None)` when all rows have been consumed.
1160 ///
1161 /// # Example
1162 ///
1163 /// ```no_run
1164 /// # use hyperdb_api::{Rowset, Result};
1165 /// # fn example(mut result: Rowset) -> Result<()> {
1166 /// while let Some(chunk) = result.next_chunk()? {
1167 /// for row in &chunk {
1168 /// let id: Option<i32> = row.get(0); // Generic typed access
1169 /// let value = row.get_f64(1); // Direct accessor
1170 /// }
1171 /// }
1172 /// # Ok(())
1173 /// # }
1174 /// ```
1175 ///
1176 /// # Errors
1177 ///
1178 /// - Returns [`crate::Error::Server`] if the server sends an `ErrorResponse`
1179 /// while streaming the result set.
1180 /// - Returns [`crate::Error::Io`] on transport-level I/O failures.
1181 /// - Returns [`crate::Error::Conversion`] if an Arrow IPC chunk cannot be decoded.
1182 pub fn next_chunk(&mut self) -> Result<Option<Vec<Row>>> {
1183 // Pull the next raw chunk from the underlying transport first;
1184 // on TCP, this is what makes the `RowDescription` bytes arrive
1185 // so we can cache the schema in the step below. We collect a
1186 // `TransportChunk` instead of a `Vec<Row>` directly so the
1187 // schema can be attached after we've populated the cache.
1188 enum TransportChunk {
1189 Tcp(Vec<StreamRow>),
1190 Arrow(Arc<RecordBatch>),
1191 }
1192
1193 let chunk_opt: Option<TransportChunk> = match &mut self.inner {
1194 RowsetInner::Tcp(stream) => stream.next_chunk()?.map(TransportChunk::Tcp),
1195 RowsetInner::Arrow(arrow) => arrow
1196 .next_chunk()?
1197 .map(|chunk| TransportChunk::Arrow(Arc::new(chunk.into_batch()))),
1198 RowsetInner::Prepared(stream) => stream.next_chunk()?.map(TransportChunk::Tcp),
1199 };
1200
1201 let Some(chunk) = chunk_opt else {
1202 return Ok(None);
1203 };
1204
1205 // Populate the schema cache if not already set, then clone the
1206 // Arc into each Row so `Row::get::<Numeric>` and friends can
1207 // look up per-column precision / scale without any caller
1208 // having to thread the schema through manually.
1209 let schema = self.cached_schema_arc();
1210 let rows = match chunk {
1211 TransportChunk::Tcp(stream_rows) => stream_rows
1212 .into_iter()
1213 .map(|row| Row::from_tcp(row, schema.clone()))
1214 .collect(),
1215 TransportChunk::Arrow(batch) => (0..batch.num_rows())
1216 .map(|row_index| Row::from_arrow(Arc::clone(&batch), row_index, schema.clone()))
1217 .collect(),
1218 };
1219 Ok(Some(rows))
1220 }
1221
1222 /// Returns an iterator over all rows in the result set.
1223 ///
1224 /// This provides a C++-like iteration experience while maintaining
1225 /// Rust's explicit error handling. Chunks are fetched internally
1226 /// as needed, keeping memory usage constant.
1227 ///
1228 /// # Example
1229 ///
1230 /// ```no_run
1231 /// # use hyperdb_api::{Connection, Result};
1232 /// # fn example(conn: &Connection) -> Result<()> {
1233 /// // Simple iteration (like C++)
1234 /// let result = conn.execute_query("SELECT * FROM users")?;
1235 /// for row in result.rows() {
1236 /// let row = row?; // Handle potential network errors
1237 /// let id: Option<i32> = row.get(0);
1238 /// let name: Option<String> = row.get(1);
1239 /// println!("User: {:?} - {:?}", id, name);
1240 /// }
1241 /// # Ok(())
1242 /// # }
1243 /// ```
1244 ///
1245 /// # Error Handling
1246 ///
1247 /// Unlike C++ which uses exceptions, Rust requires explicit error handling.
1248 /// Each item in the iterator is a `Result<LightweightRow>` to handle
1249 /// potential network or protocol errors during streaming.
1250 ///
1251 /// # Comparison with `next_chunk()`
1252 ///
1253 /// | Aspect | `rows()` | `next_chunk()` |
1254 /// |--------|----------|----------------|
1255 /// | Syntax | Simpler, C++-like | More verbose |
1256 /// | Error handling | Per-row with `?` | Per-chunk |
1257 /// | Batch ops | Use `.collect()` | Natural |
1258 /// | Best for | Simple iteration | Batch processing |
1259 #[must_use]
1260 pub fn rows(self) -> RowIterator<'conn> {
1261 RowIterator {
1262 rowset: self,
1263 current_iter: Vec::new().into_iter(),
1264 }
1265 }
1266
1267 /// Collects all rows into a Vec.
1268 ///
1269 /// This is a convenience method that handles error collection more elegantly
1270 /// than the standard `collect::<Result<Vec<_>, _>>()` pattern.
1271 ///
1272 /// # Example
1273 ///
1274 /// ```no_run
1275 /// # use hyperdb_api::{Connection, Result};
1276 /// # fn example(conn: &Connection) -> Result<()> {
1277 /// let result = conn.execute_query("SELECT id, name FROM users")?;
1278 /// let rows = result.collect_rows()?; // Much cleaner than collect::<Result<Vec<_>, _>>()
1279 ///
1280 /// for row in rows {
1281 /// let id: Option<i32> = row.get(0);
1282 /// let name: Option<String> = row.get(1);
1283 /// println!("User: {:?} - {:?}", id, name);
1284 /// }
1285 /// # Ok(())
1286 /// # }
1287 /// ```
1288 ///
1289 /// # Errors
1290 ///
1291 /// Returns the first error produced by [`next_chunk`](Self::next_chunk)
1292 /// while draining the stream (transport I/O failure or server-side
1293 /// error).
1294 pub fn collect_rows(self) -> crate::error::Result<Vec<Row>> {
1295 self.rows().collect::<crate::error::Result<Vec<_>>>()
1296 }
1297
1298 /// Collects the first column of each row into a Vec.
1299 ///
1300 /// This is useful for single-column queries or when you only need one column.
1301 ///
1302 /// # Example
1303 ///
1304 /// ```no_run
1305 /// # use hyperdb_api::{Connection, Result};
1306 /// # fn example(conn: &Connection) -> Result<()> {
1307 /// let result = conn.execute_query("SELECT name FROM users")?;
1308 /// let names: Vec<Option<String>> = result.collect_column()?;
1309 ///
1310 /// for name in names {
1311 /// if let Some(name) = name {
1312 /// println!("User: {}", name);
1313 /// }
1314 /// }
1315 /// # Ok(())
1316 /// # }
1317 /// ```
1318 ///
1319 /// # Errors
1320 ///
1321 /// Returns the first streaming error from
1322 /// [`next_chunk`](Self::next_chunk). SQL `NULL` cells yield
1323 /// `Option::None` entries, not errors.
1324 pub fn collect_column<T: crate::result::RowValue>(
1325 self,
1326 ) -> crate::error::Result<Vec<Option<T>>> {
1327 self.rows()
1328 .map(|row| row.map(|r| r.get::<T>(0)))
1329 .collect::<crate::error::Result<Vec<_>>>()
1330 }
1331
1332 /// Collects the first column, filtering out NULL values.
1333 ///
1334 /// This is useful when you know the column doesn't contain NULLs or want to ignore them.
1335 ///
1336 /// # Example
1337 ///
1338 /// ```no_run
1339 /// # use hyperdb_api::{Connection, Result};
1340 /// # fn example(conn: &Connection) -> Result<()> {
1341 /// let result = conn.execute_query("SELECT name FROM users WHERE name IS NOT NULL")?;
1342 /// let names: Vec<String> = result.collect_column_non_null()?;
1343 ///
1344 /// for name in names {
1345 /// println!("User: {}", name); // No need to handle Option
1346 /// }
1347 /// # Ok(())
1348 /// # }
1349 /// ```
1350 ///
1351 /// # Errors
1352 ///
1353 /// Returns the first streaming error from
1354 /// [`collect_column`](Self::collect_column).
1355 pub fn collect_column_non_null<T: crate::result::RowValue>(
1356 self,
1357 ) -> crate::error::Result<Vec<T>> {
1358 Ok(self.collect_column::<T>()?.into_iter().flatten().collect())
1359 }
1360
1361 /// Gets the first row of the result set.
1362 ///
1363 /// This is useful for queries that are expected to return exactly one row,
1364 /// such as aggregate queries or lookups by unique key.
1365 ///
1366 /// # Example
1367 ///
1368 /// ```no_run
1369 /// # use hyperdb_api::{Connection, Result};
1370 /// # fn example(conn: &Connection) -> Result<()> {
1371 /// let result = conn.execute_query("SELECT COUNT(*) FROM users")?;
1372 /// if let Some(row) = result.first_row()? {
1373 /// let count: Option<i64> = row.get(0);
1374 /// println!("User count: {:?}", count);
1375 /// }
1376 /// # Ok(())
1377 /// # }
1378 /// ```
1379 ///
1380 /// # Errors
1381 ///
1382 /// Returns the error from [`next_chunk`](Self::next_chunk). An empty
1383 /// result set yields `Ok(None)`, not an error.
1384 pub fn first_row(mut self) -> crate::error::Result<Option<Row>> {
1385 if let Some(chunk) = self.next_chunk()? {
1386 Ok(chunk.into_iter().next())
1387 } else {
1388 Ok(None)
1389 }
1390 }
1391
1392 /// Gets the first row or returns an error if no rows were found.
1393 ///
1394 /// This is useful when you expect exactly one row and want to fail if that's not the case.
1395 ///
1396 /// # Example
1397 ///
1398 /// ```no_run
1399 /// # use hyperdb_api::{Connection, Result};
1400 /// # fn example(conn: &Connection) -> Result<()> {
1401 /// let result = conn.execute_query("SELECT id, name FROM users WHERE id = 1")?;
1402 /// let row = result.require_first_row()?; // Fails if no row found
1403 /// let id: Option<i32> = row.get(0);
1404 /// let name: Option<String> = row.get(1);
1405 /// println!("Found user: {:?} - {:?}", id, name);
1406 /// # Ok(())
1407 /// # }
1408 /// ```
1409 ///
1410 /// # Errors
1411 ///
1412 /// - Returns the error from [`first_row`](Self::first_row).
1413 /// - Returns [`crate::Error::Conversion`] with message `"Query returned no rows"`
1414 /// if the result set is empty.
1415 pub fn require_first_row(self) -> crate::error::Result<Row> {
1416 self.first_row()?
1417 .ok_or_else(|| crate::error::Error::conversion("Query returned no rows"))
1418 }
1419
1420 /// Gets a scalar value from the first row, first column.
1421 ///
1422 /// This is a convenience method for scalar queries like `SELECT COUNT(*)` or `SELECT MAX(id)`.
1423 ///
1424 /// # Example
1425 ///
1426 /// ```no_run
1427 /// # use hyperdb_api::{Connection, Result};
1428 /// # fn example(conn: &Connection) -> Result<()> {
1429 /// let result = conn.execute_query("SELECT COUNT(*) FROM users")?;
1430 /// let count: Option<i64> = result.scalar()?; // Much cleaner than manual row handling
1431 /// println!("User count: {:?}", count);
1432 /// # Ok(())
1433 /// # }
1434 /// ```
1435 ///
1436 /// # Errors
1437 ///
1438 /// Returns the error from [`require_first_row`](Self::require_first_row):
1439 /// streaming error or empty result. SQL `NULL` in the single cell
1440 /// yields `Ok(None)`.
1441 pub fn scalar<T: crate::result::RowValue>(self) -> crate::error::Result<Option<T>> {
1442 Ok(self.require_first_row()?.get(0))
1443 }
1444
1445 /// Gets a scalar value from the first row, first column, or returns an error if NULL.
1446 ///
1447 /// This is useful when you expect a non-NULL scalar result.
1448 ///
1449 /// # Example
1450 ///
1451 /// ```no_run
1452 /// # use hyperdb_api::{Connection, Result};
1453 /// # fn example(conn: &Connection) -> Result<()> {
1454 /// let result = conn.execute_query("SELECT COUNT(*) FROM users")?;
1455 /// let count: i64 = result.require_scalar()?; // Fails if NULL
1456 /// println!("User count: {}", count);
1457 /// # Ok(())
1458 /// # }
1459 /// ```
1460 ///
1461 /// # Errors
1462 ///
1463 /// - Returns the error from [`scalar`](Self::scalar).
1464 /// - Returns [`crate::Error::Conversion`] with message `"Scalar query returned NULL"`
1465 /// if the single cell is SQL `NULL`.
1466 pub fn require_scalar<T: crate::result::RowValue>(self) -> crate::error::Result<T> {
1467 self.scalar()?
1468 .ok_or_else(|| crate::error::Error::conversion("Scalar query returned NULL"))
1469 }
1470}
1471
1472// =============================================================================
1473// RowIterator - C++-like iteration over query results
1474// =============================================================================
1475
1476/// An iterator over rows in a query result set.
1477///
1478/// `RowIterator` provides a C++-like iteration experience, hiding the
1479/// chunked fetching internally. Each call to `next()` returns the next
1480/// row, automatically fetching new chunks as needed.
1481///
1482/// # Memory Behavior
1483///
1484/// Memory usage remains constant regardless of result set size:
1485/// - Internally fetches 64K rows at a time
1486/// - Previous chunks are dropped when exhausted
1487/// - Safe for billion-row results
1488///
1489/// # Example
1490///
1491/// ```no_run
1492/// # use hyperdb_api::{Connection, Result};
1493/// # fn example(conn: &Connection) -> Result<()> {
1494/// let result = conn.execute_query("SELECT id, name FROM users")?;
1495/// for row in result.rows() {
1496/// let row = row?;
1497/// let id = row.get_i32(0).unwrap_or(-1);
1498/// let name = row.get::<String>(1).unwrap_or_default();
1499/// println!("{}: {}", id, name);
1500/// }
1501/// # Ok(())
1502/// # }
1503/// ```
1504///
1505/// # Error Handling
1506///
1507/// Each iteration yields a `Result<Row>`. Errors can occur
1508/// when fetching new chunks from the server (network issues, protocol
1509/// errors, etc.). Use `?` or match to handle them:
1510///
1511/// ```no_run
1512/// # use hyperdb_api::{Rowset, Result};
1513/// # fn example(mut result: Rowset) -> Result<()> {
1514/// // Using ? in a function that returns Result
1515/// for row in result.rows() {
1516/// let row = row?;
1517/// // process row...
1518/// }
1519/// # Ok(())
1520/// # }
1521/// # fn example2(mut result: Rowset) -> Result<()> {
1522/// // Using try_for_each
1523/// result.rows().try_for_each(|row| -> Result<()> {
1524/// let row = row?;
1525/// // process row...
1526/// Ok(())
1527/// })?;
1528/// # Ok(())
1529/// # }
1530/// ```
1531#[derive(Debug)]
1532pub struct RowIterator<'conn> {
1533 rowset: Rowset<'conn>,
1534 current_iter: std::vec::IntoIter<Row>,
1535}
1536
1537impl Iterator for RowIterator<'_> {
1538 type Item = Result<Row>;
1539
1540 fn next(&mut self) -> Option<Self::Item> {
1541 // Try to get next row from current chunk
1542 if let Some(row) = self.current_iter.next() {
1543 return Some(Ok(row));
1544 }
1545
1546 // Current chunk exhausted, fetch next chunk
1547 match self.rowset.next_chunk() {
1548 Ok(Some(chunk)) => {
1549 self.current_iter = chunk.into_iter();
1550 // Return first row of new chunk
1551 self.current_iter.next().map(Ok)
1552 }
1553 Ok(None) => None, // No more rows
1554 Err(e) => Some(Err(e)), // Error fetching chunk
1555 }
1556 }
1557}
1558
1559// =============================================================================
1560// Unit tests that don't need a live hyperd backend.
1561//
1562// Anything requiring a real Hyper process lives in `hyperdb-api/tests/*.rs` where
1563// `TestConnection` spins up a `HyperProcess` per test. These tests exercise
1564// pure in-process logic — specifically the Arrow-path branches of
1565// `Row::get_numeric`, where we can construct a synthetic `RecordBatch` with a
1566// specific `DataType::Decimal128(p, s)` descriptor and probe `Row`'s
1567// handling of it without hyperd in the loop.
1568// =============================================================================
1569
1570#[cfg(test)]
1571mod arrow_path_tests {
1572 use super::*;
1573 use arrow::array::Decimal128Array;
1574 use arrow::datatypes::{DataType as ArrowType, Field, Schema};
1575
1576 /// Build a single-row `RecordBatch` with a Decimal128 column whose
1577 /// value is `raw` and whose precision/scale are those passed in.
1578 fn decimal128_batch(raw: i128, precision: u8, scale: i8) -> Arc<RecordBatch> {
1579 let array = Decimal128Array::from(vec![Some(raw)])
1580 .with_precision_and_scale(precision, scale)
1581 .expect("valid Arrow Decimal128");
1582 let field = Field::new("v", ArrowType::Decimal128(precision, scale), true);
1583 let schema = Arc::new(Schema::new(vec![field]));
1584 Arc::new(RecordBatch::try_new(schema, vec![Arc::new(array)]).expect("batch"))
1585 }
1586
1587 /// Happy-path: a positive-scale Arrow Decimal128 decodes correctly
1588 /// via `row.get::<Numeric>()`, locking in the common case alongside
1589 /// the negative-scale test below.
1590 #[test]
1591 fn get_numeric_reads_arrow_decimal128_with_positive_scale() {
1592 // NUMERIC(10, 2), unscaled value 123 → 1.23
1593 let batch = decimal128_batch(123, 10, 2);
1594 let row = Row::from_arrow(Arc::clone(&batch), 0, None);
1595
1596 let numeric = row.get_numeric(0).expect("Some for positive-scale decimal");
1597 assert_eq!(numeric.unscaled_value(), 123);
1598 assert_eq!(numeric.scale(), 2);
1599 assert!((numeric.to_f64() - 1.23).abs() < 1e-9);
1600
1601 // Same result via the generic `row.get::<Numeric>` path.
1602 let via_rowvalue: hyperdb_api_core::types::Numeric =
1603 row.get(0).expect("RowValue path agrees with get_numeric");
1604 assert_eq!(via_rowvalue, numeric);
1605 }
1606
1607 /// Arrow's `DataType::Decimal128(u8, i8)` allows negative scale —
1608 /// a legitimate Arrow concept meaning "raw × 10^abs(scale)" (e.g.
1609 /// scale=-2 with raw=5 renders as 500). Hyper's `Numeric` uses a
1610 /// `u8` scale with no representation for that multiplier.
1611 ///
1612 /// The earlier `.max(0) as u8` code silently clamped the scale to
1613 /// 0 while keeping `raw` unchanged — which produces a value with
1614 /// the wrong magnitude (`5` instead of `500` in the example
1615 /// above). The fix here is to reject negative scales via
1616 /// `try_into` + `?`, which surfaces as `None` to the caller.
1617 /// That's strictly safer than a silent-wrong-magnitude value.
1618 #[test]
1619 fn get_numeric_rejects_arrow_decimal128_with_negative_scale() {
1620 // NUMERIC(10, -2) — Arrow allows this; Hyper's Numeric can't
1621 // represent it. Our `get_numeric` must return None rather
1622 // than silently drop the negative-scale multiplier.
1623 let batch = decimal128_batch(5, 10, -2);
1624 let row = Row::from_arrow(Arc::clone(&batch), 0, None);
1625
1626 assert!(
1627 row.get_numeric(0).is_none(),
1628 "negative Arrow scale must not produce a silently-wrong-magnitude Numeric",
1629 );
1630
1631 // And the same through the `RowValue` blanket path.
1632 let via_rowvalue: Option<hyperdb_api_core::types::Numeric> = row.get(0);
1633 assert!(via_rowvalue.is_none());
1634 }
1635
1636 /// Boundary: scale = 0 is a legal `u8` and must still succeed.
1637 /// Guards against an over-tightened check that accidentally
1638 /// rejects zero along with negatives.
1639 #[test]
1640 fn get_numeric_accepts_arrow_decimal128_with_zero_scale() {
1641 let batch = decimal128_batch(42, 10, 0);
1642 let row = Row::from_arrow(Arc::clone(&batch), 0, None);
1643 let numeric = row.get_numeric(0).expect("scale 0 is fine");
1644 assert_eq!(numeric.unscaled_value(), 42);
1645 assert_eq!(numeric.scale(), 0);
1646 }
1647}