exarrow-rs 0.28.0

//! Result set handling and iteration.
//!
//! This module provides types for handling query results, including
//! streaming result sets and metadata.

use crate::error::{ColumnConversionError, ConversionError, QueryError};
use crate::transport::messages::{ColumnInfo, ResultData, ResultPayload, ResultSetHandle};
use crate::transport::protocol::QueryResult as TransportQueryResult;
use crate::transport::TransportProtocol;
use crate::types::conversion::parse_interval;
use crate::types::{ColumnMetadata, ExasolType, SchemaBuilder};
use arrow::array::{
    new_empty_array, Array, BinaryArray, BinaryBuilder, BooleanArray, BooleanBuilder,
    Decimal128Array, Decimal128Builder, IntervalMonthDayNanoArray, IntervalMonthDayNanoBuilder,
    PrimitiveArray, PrimitiveBuilder, RecordBatch, RecordBatchOptions, StringArray, StringBuilder,
};
use arrow::datatypes::{
    ArrowPrimitiveType, DataType, Date32Type, Field, Float64Type, Int32Type, Int64Type,
    IntervalUnit, Schema, TimestampMicrosecondType,
};
use serde_json::Value;
use std::num::{IntErrorKind, ParseIntError};
use std::sync::Arc;
use tokio::sync::Mutex;

/// Metadata about a query execution.
#[derive(Debug, Clone)]
pub struct QueryMetadata {
    /// Schema of the result set
    pub schema: Arc<Schema>,
    /// Total number of rows (if known)
    pub total_rows: Option<i64>,
    /// Number of columns
    pub column_count: usize,
    /// Execution time in milliseconds (if available)
    pub execution_time_ms: Option<u64>,
}

impl QueryMetadata {
    /// Create metadata from schema and row count.
    pub fn new(schema: Arc<Schema>, total_rows: Option<i64>) -> Self {
        Self {
            column_count: schema.fields().len(),
            schema,
            total_rows,
            execution_time_ms: None,
        }
    }

    /// Set execution time.
    pub fn with_execution_time(mut self, execution_time_ms: u64) -> Self {
        self.execution_time_ms = Some(execution_time_ms);
        self
    }

    /// Get column names.
    pub fn column_names(&self) -> Vec<&str> {
        self.schema
            .fields()
            .iter()
            .map(|f| f.name().as_str())
            .collect()
    }

    /// Get column types.
    pub fn column_types(&self) -> Vec<&arrow::datatypes::DataType> {
        self.schema.fields().iter().map(|f| f.data_type()).collect()
    }
}

/// Query result set that can be either a row count or streaming data.
pub struct ResultSet {
    /// Result type
    inner: ResultSetInner,
    /// Transport reference for fetching more data
    transport: Arc<Mutex<dyn TransportProtocol>>,
}

impl std::fmt::Debug for ResultSet {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ResultSet")
            .field("inner", &self.inner)
            .field("transport", &"<TransportProtocol>")
            .finish()
    }
}

#[derive(Debug)]
enum ResultSetInner {
    /// Row count result (INSERT, UPDATE, DELETE)
    RowCount { count: i64 },
    /// Streaming result set (SELECT)
    Stream {
        /// Result set handle for fetching
        handle: Option<ResultSetHandle>,
        /// Query metadata
        metadata: QueryMetadata,
        /// Buffered batches
        batches: Vec<RecordBatch>,
        /// Whether all data has been fetched
        complete: bool,
    },
}

/// Resolve the Exasol type that a transport column description names.
pub(crate) fn exasol_type_of(
    data_type: &crate::transport::messages::DataType,
) -> Result<ExasolType, ConversionError> {
    let exasol_type = match data_type.type_name.as_str() {
        "BOOLEAN" => ExasolType::Boolean,
        "CHAR" => ExasolType::Char {
            size: data_type.size.unwrap_or(1) as usize,
        },
        "VARCHAR" => ExasolType::Varchar {
            size: data_type.size.unwrap_or(2000000) as usize,
        },
        "DECIMAL" => ExasolType::Decimal {
            precision: data_type.precision.unwrap_or(18) as u8,
            scale: data_type.scale.unwrap_or(0) as i8,
        },
        "DOUBLE" => ExasolType::Double,
        "DATE" => ExasolType::Date,
        "TIMESTAMP" => ExasolType::Timestamp {
            with_local_time_zone: data_type.with_local_time_zone.unwrap_or(false),
        },
        "TIMESTAMP WITH LOCAL TIME ZONE" => ExasolType::Timestamp {
            with_local_time_zone: true,
        },
        "INTERVAL YEAR TO MONTH" => ExasolType::IntervalYearToMonth,
        "INTERVAL DAY TO SECOND" => ExasolType::IntervalDayToSecond {
            precision: data_type.fraction.unwrap_or(3) as u8,
        },
        "GEOMETRY" => ExasolType::Geometry {
            srid: data_type.srid,
        },
        "HASHTYPE" => ExasolType::hashtype_from_reported_size(data_type.size),
        _ => {
            return Err(ConversionError::UnsupportedType {
                exasol_type: data_type.type_name.clone(),
            })
        }
    };

    Ok(exasol_type)
}

/// Why `ResultSet::fetch_page_after` failed. The iterator ends after a row
/// count mismatch, because the result set cannot be read further.
enum PageFetchError {
    Fetch(QueryError),
    RowCountMismatch(String),
}

impl From<PageFetchError> for QueryError {
    fn from(error: PageFetchError) -> Self {
        match error {
            PageFetchError::Fetch(error) => error,
            PageFetchError::RowCountMismatch(message) => QueryError::ExecutionFailed(message),
        }
    }
}

impl ResultSet {
    /// Create a result set from a transport query result.
    pub(crate) fn from_transport_result(
        result: TransportQueryResult,
        transport: Arc<Mutex<dyn TransportProtocol>>,
    ) -> Result<Self, QueryError> {
        match result {
            TransportQueryResult::RowCount { count } => Ok(Self {
                inner: ResultSetInner::RowCount { count },
                transport,
            }),
            TransportQueryResult::ResultSet { handle, data } => {
                let schema = Self::build_schema(&data.columns)
                    .map_err(|e| QueryError::ExecutionFailed(e.to_string()))?;

                let metadata = QueryMetadata::new(Arc::clone(&schema), Some(data.total_rows));

                let num_rows_received = data.data.num_rows() as i64;

                // A result set always carries column metadata, even with zero rows.
                // Emit one (possibly empty) batch built from that schema so consumers
                // that read the schema from the first batch (e.g. ADBC's
                // RecordBatchReader, used by dbt Fusion for `WHERE FALSE LIMIT 0`
                // schema probes) still receive the columns. Returning an empty Vec
                // here would drop the schema for zero-row results.
                let batch = if data.data.is_empty() {
                    Self::empty_record_batch(&schema)
                } else {
                    Self::payload_to_record_batch(&data, &schema)
                }
                .map_err(|e| QueryError::ExecutionFailed(e.to_string()))?;
                let batches = vec![batch];

                let complete = handle.is_none() || data.total_rows == num_rows_received;

                Ok(Self {
                    inner: ResultSetInner::Stream {
                        handle, // Already Option<ResultSetHandle>
                        metadata,
                        batches,
                        complete,
                    },
                    transport,
                })
            }
        }
    }

    /// Get the row count if this is a row count result.
    pub fn row_count(&self) -> Option<i64> {
        match &self.inner {
            ResultSetInner::RowCount { count } => Some(*count),
            _ => None,
        }
    }

    /// Get the metadata if this is a streaming result.
    pub fn metadata(&self) -> Option<&QueryMetadata> {
        match &self.inner {
            ResultSetInner::Stream { metadata, .. } => Some(metadata),
            _ => None,
        }
    }

    /// Check if this is a streaming result set.
    pub fn is_stream(&self) -> bool {
        matches!(&self.inner, ResultSetInner::Stream { .. })
    }

    /// Convert to an iterator over RecordBatches.
    ///
    /// # Errors
    /// Returns `QueryError::NoResultSet` if this is not a streaming result.
    pub fn into_iterator(self) -> Result<ResultSetIterator, QueryError> {
        match self.inner {
            ResultSetInner::Stream {
                handle,
                metadata,
                batches,
                complete,
            } => Ok(ResultSetIterator {
                handle,
                transport: self.transport,
                metadata,
                batches,
                current_index: 0,
                complete,
            }),
            ResultSetInner::RowCount { .. } => Err(QueryError::NoResultSet(
                "Cannot iterate over row count result".to_string(),
            )),
        }
    }

    /// Fetch all remaining batches into memory.
    ///
    /// # Errors
    /// Returns `QueryError` if fetching fails or if this is not a streaming result.
    /// Returns `QueryError::ExecutionFailed` when the result set delivers fewer or
    /// more rows than its total row count.
    pub async fn fetch_all(mut self) -> Result<Vec<RecordBatch>, QueryError> {
        let ResultSetInner::Stream {
            handle,
            metadata,
            batches,
            complete,
        } = &mut self.inner
        else {
            return Err(QueryError::NoResultSet(
                "Cannot fetch batches from row count result".to_string(),
            ));
        };

        let pagination = match *handle {
            Some(handle_val) if !*complete => {
                Self::paginate_remaining(&self.transport, handle_val, metadata, batches.clone())
                    .await
            }
            _ => Self::check_buffered_rows(metadata, batches).map(|()| batches.clone()),
        };

        // Close the result set handle on the server to release resources
        if let Some(handle_val) = handle.take() {
            let mut transport = self.transport.lock().await;
            // Ignore close errors - the pagination result is what the caller needs
            let _ = transport.close_result_set(handle_val).await;
        }

        pagination
    }

    /// Fetch every remaining page of a result set, appending to `collected`.
    ///
    /// The per-page `total_rows` reported by the transport is deliberately
    /// ignored: it describes that page, so trusting it would end pagination
    /// early. The end rule lives in `fetch_page_after`.
    async fn paginate_remaining(
        transport: &Arc<Mutex<dyn TransportProtocol>>,
        handle: ResultSetHandle,
        metadata: &QueryMetadata,
        mut collected: Vec<RecordBatch>,
    ) -> Result<Vec<RecordBatch>, QueryError> {
        while let Some(batch) =
            Self::fetch_page_after(transport, handle, metadata, Self::row_count_of(&collected))
                .await
                .map_err(QueryError::from)?
        {
            collected.push(batch);
        }
        Ok(collected)
    }

    /// Fetch the page that follows `rows_received` rows, or `None` at the end.
    ///
    /// `fetch_all` and `ResultSetIterator` share this end-of-stream rule. When
    /// the query metadata reports a total above zero, the stream ends exactly
    /// when the rows received reach it, with no further fetch. A page that
    /// ends the stream early or overshoots the total is an error, because
    /// Exasol reports the exact total with every result set. A total of zero
    /// means the count is unknown, so the first empty page ends the stream.
    async fn fetch_page_after(
        transport: &Arc<Mutex<dyn TransportProtocol>>,
        handle: ResultSetHandle,
        metadata: &QueryMetadata,
        rows_received: usize,
    ) -> Result<Option<RecordBatch>, PageFetchError> {
        let total = metadata.total_rows.unwrap_or(0).max(0) as usize;
        if total > 0 && rows_received >= total {
            return Ok(None);
        }

        let result_data = transport
            .lock()
            .await
            .fetch_results(handle)
            .await
            .map_err(|e| PageFetchError::Fetch(QueryError::ExecutionFailed(e.to_string())))?;

        if result_data.data.is_empty() {
            if total == 0 {
                return Ok(None);
            }
            return Err(PageFetchError::RowCountMismatch(format!(
                "Result set ended after {rows_received} rows, but its total row count is {total}"
            )));
        }

        let batch = Self::payload_to_record_batch(&result_data, &metadata.schema)
            .map_err(|e| PageFetchError::Fetch(QueryError::ExecutionFailed(e.to_string())))?;

        let rows_after_page = rows_received + batch.num_rows();
        if total > 0 && rows_after_page > total {
            return Err(PageFetchError::RowCountMismatch(format!(
                "Result set delivered {rows_after_page} rows, but its total row count is {total}"
            )));
        }

        Ok(Some(batch))
    }

    /// Verify that buffered rows match a known total, for results that need no fetch.
    fn check_buffered_rows(
        metadata: &QueryMetadata,
        batches: &[RecordBatch],
    ) -> Result<(), QueryError> {
        let total = metadata.total_rows.unwrap_or(0).max(0) as usize;
        let rows = Self::row_count_of(batches);
        if total > 0 && rows != total {
            return Err(QueryError::ExecutionFailed(format!(
                "Result set delivered {rows} rows, but its total row count is {total}"
            )));
        }
        Ok(())
    }

    fn row_count_of(batches: &[RecordBatch]) -> usize {
        batches.iter().map(|b| b.num_rows()).sum()
    }

    /// Build the result set metadata schema from column information.
    ///
    /// Query results and ADBC `execute_schema` must report the same column
    /// types, so both build the schema with this function. Every field is
    /// nullable, because Exasol reports no column nullability, and carries the
    /// `exasol:` field metadata of its column.
    pub(crate) fn build_schema(columns: &[ColumnInfo]) -> Result<Arc<Schema>, ConversionError> {
        let column_metadata = columns
            .iter()
            .map(|col| {
                Ok(ColumnMetadata {
                    name: col.name.clone(),
                    data_type: exasol_type_of(&col.data_type)?,
                    nullable: true,
                })
            })
            .collect::<Result<Vec<_>, ConversionError>>()?;

        Ok(Arc::new(
            SchemaBuilder::new().add_columns(column_metadata).build()?,
        ))
    }

    /// Convert a ResultData payload to RecordBatch, handling both JSON and Arrow variants.
    ///
    /// An Arrow batch keeps its own column arrays, field names, Arrow data types,
    /// and nullability, and takes the field metadata of the result set metadata
    /// schema. The native transport can decode a column to another Arrow type
    /// than `TypeMapper` maps: a `T_TIMESTAMP_LOCAL_TZ` column, which Exasol
    /// 2025.2.1 does not send, would decode to Timestamp(Microsecond, None) under
    /// a Timestamp(Microsecond, "UTC") field. So the metadata goes onto the
    /// batch's own fields.
    fn payload_to_record_batch(
        data: &ResultData,
        schema: &Arc<Schema>,
    ) -> Result<RecordBatch, ConversionError> {
        match &data.data {
            ResultPayload::Arrow(batch) => Self::with_field_metadata(batch, schema),
            ResultPayload::Json(rows) => Self::column_major_to_record_batch(rows, schema),
        }
    }

    fn with_field_metadata(
        batch: &RecordBatch,
        schema: &Schema,
    ) -> Result<RecordBatch, ConversionError> {
        let batch_schema = batch.schema();
        if batch_schema.fields().len() != schema.fields().len() {
            return Err(ConversionError::SchemaMismatch(format!(
                "Arrow batch has {} columns, but the result set metadata schema has {}",
                batch_schema.fields().len(),
                schema.fields().len()
            )));
        }

        let fields: Vec<Field> = batch_schema
            .fields()
            .iter()
            .zip(schema.fields())
            .map(|(batch_field, schema_field)| {
                batch_field
                    .as_ref()
                    .clone()
                    .with_metadata(schema_field.metadata().clone())
            })
            .collect();
        let schema_with_metadata =
            Schema::new_with_metadata(fields, batch_schema.metadata().clone());

        let options = RecordBatchOptions::new().with_row_count(Some(batch.num_rows()));
        RecordBatch::try_new_with_options(
            Arc::new(schema_with_metadata),
            batch.columns().to_vec(),
            &options,
        )
        .map_err(|e| ConversionError::ArrowError(e.to_string()))
    }

    /// Build a zero-row RecordBatch that carries the given schema.
    ///
    /// Used for empty result sets so the column schema is still conveyed to
    /// consumers that read it from the first batch (the schema itself comes
    /// from the result set's column metadata, which is present even with no
    /// rows).
    fn empty_record_batch(schema: &Arc<Schema>) -> Result<RecordBatch, ConversionError> {
        let arrays: Vec<Arc<dyn Array>> = schema
            .fields()
            .iter()
            .map(|field| new_empty_array(field.data_type()))
            .collect();
        RecordBatch::try_new(Arc::clone(schema), arrays)
            .map_err(|e| ConversionError::ArrowError(e.to_string()))
    }

    /// Convert row-major JSON result data to RecordBatch.
    fn column_major_to_record_batch(
        rows: &[Vec<Value>],
        schema: &Arc<Schema>,
    ) -> Result<RecordBatch, ConversionError> {
        if rows.is_empty() {
            return Self::empty_record_batch(schema);
        }

        let arrays: Vec<Arc<dyn Array>> = schema
            .fields()
            .iter()
            .enumerate()
            .map(|(column_index, field)| {
                let values = Self::column_of(rows, column_index);
                Self::json_column_to_array(field, &values).map_err(|error| {
                    error.into_conversion_error(
                        column_index,
                        field.name(),
                        &values,
                        field.data_type(),
                    )
                })
            })
            .collect::<Result<_, _>>()?;

        RecordBatch::try_new(Arc::clone(schema), arrays)
            .map_err(|e| ConversionError::ArrowError(e.to_string()))
    }

    /// Slice one column out of row-major JSON rows, padding short rows with NULL.
    fn column_of(rows: &[Vec<Value>], column_index: usize) -> Vec<&Value> {
        rows.iter()
            .map(|row| row.get(column_index).unwrap_or(&Value::Null))
            .collect()
    }

    /// Build the Arrow array for one result column from its JSON values.
    ///
    /// A value that does not fit the target Arrow type becomes NULL instead of
    /// failing the batch: Exasol renders the same logical type differently
    /// depending on transport and column width, so a single unexpected shape
    /// must not discard an otherwise valid result set. Only a structurally
    /// impossible column is reported as an error: an out-of-range DECIMAL
    /// precision/scale, a decimal string that cannot be scaled, an INTERVAL
    /// value that is not a string `parse_interval` reads (including a year
    /// to month value beyond the Arrow month range), or an Arrow type that no
    /// result column maps to.
    fn json_column_to_array(
        field: &Field,
        values: &[&Value],
    ) -> Result<Arc<dyn Array>, ColumnConversionError> {
        Ok(match field.data_type() {
            DataType::Boolean => Arc::new(Self::json_to_boolean_array(values)),
            DataType::Int32 => Arc::new(Self::json_to_primitive_array::<Int32Type, _>(
                values,
                |value| value.as_i64().map(|i| i as i32),
            )),
            DataType::Int64 => Arc::new(Self::json_to_primitive_array::<Int64Type, _>(
                values,
                |value| value.as_i64(),
            )),
            DataType::Float64 => Arc::new(Self::json_to_primitive_array::<Float64Type, _>(
                values,
                |value| value.as_f64(),
            )),
            DataType::Utf8 => Arc::new(Self::json_to_string_array(values, Self::render_as_text)),
            DataType::Decimal128(precision, scale) => {
                Arc::new(Self::json_to_decimal_array(values, *precision, *scale)?)
            }
            DataType::Date32 => Arc::new(Self::json_to_primitive_array::<Date32Type, _>(
                values,
                |value| {
                    value
                        .as_str()
                        .and_then(|s| crate::types::conversion::parse_date_to_days(s).ok())
                },
            )),
            DataType::Timestamp(_, timezone) => {
                let array =
                    Self::json_to_primitive_array::<TimestampMicrosecondType, _>(values, |value| {
                        value.as_str().and_then(|s| {
                            crate::types::conversion::parse_timestamp_to_micros(s).ok()
                        })
                    });
                match timezone {
                    Some(_) => Arc::new(array.with_timezone("UTC")),
                    None => Arc::new(array),
                }
            }
            DataType::Interval(IntervalUnit::MonthDayNano) => {
                Arc::new(Self::json_to_interval_array(values)?)
            }
            DataType::Binary => Arc::new(Self::json_to_binary_array(values)),
            other => {
                return Err(ConversionError::SchemaMismatch(format!(
                    "Column {} has Arrow type {other}, which no result set column maps to",
                    field.name()
                ))
                .into())
            }
        })
    }

    /// Build a primitive Arrow array, treating any value `extract` rejects as NULL.
    fn json_to_primitive_array<T, F>(values: &[&Value], extract: F) -> PrimitiveArray<T>
    where
        T: ArrowPrimitiveType,
        F: Fn(&Value) -> Option<T::Native>,
    {
        let mut builder = PrimitiveBuilder::<T>::with_capacity(values.len());
        for value in values {
            builder.append_option(extract(value));
        }
        builder.finish()
    }

    /// Build a boolean Arrow array, treating any non-boolean value as NULL.
    fn json_to_boolean_array(values: &[&Value]) -> BooleanArray {
        let mut builder = BooleanBuilder::with_capacity(values.len());
        for value in values {
            builder.append_option(value.as_bool());
        }
        builder.finish()
    }

    /// Build an INTERVAL array from the text Exasol sends, failing on a value that is not an INTERVAL string.
    fn json_to_interval_array(
        values: &[&Value],
    ) -> Result<IntervalMonthDayNanoArray, ColumnConversionError> {
        let mut builder = IntervalMonthDayNanoBuilder::with_capacity(values.len());
        for (row, value) in values.iter().enumerate() {
            match value {
                Value::Null => builder.append_null(),
                Value::String(text) => builder.append_value(
                    parse_interval(text)
                        .map_err(|reason| ColumnConversionError::invalid_value(row, reason))?,
                ),
                other => {
                    return Err(ColumnConversionError::invalid_value(
                        row,
                        format!("Expected an INTERVAL string, got {other}"),
                    ))
                }
            }
        }
        Ok(builder.finish())
    }

    /// Build a Binary array from the text bytes of each JSON string, treating any other value as NULL.
    fn json_to_binary_array(values: &[&Value]) -> BinaryArray {
        let mut builder = BinaryBuilder::new();
        for value in values {
            builder.append_option(value.as_str().map(str::as_bytes));
        }
        builder.finish()
    }

    /// Build a UTF-8 Arrow array, rendering every non-NULL value via `render`.
    fn json_to_string_array<F>(values: &[&Value], render: F) -> StringArray
    where
        F: Fn(&Value) -> String,
    {
        let mut builder = StringBuilder::new();
        for value in values {
            if value.is_null() {
                builder.append_null();
            } else {
                builder.append_value(render(value));
            }
        }
        builder.finish()
    }

    /// Render a value destined for a VARCHAR/CHAR column.
    ///
    /// JSON strings are taken verbatim; anything else falls back to its JSON
    /// text form so that a mistyped column still round-trips its content.
    fn render_as_text(value: &Value) -> String {
        value
            .as_str()
            .map_or_else(|| value.to_string(), String::from)
    }

    /// Build a DECIMAL Arrow array at the column's declared precision and scale.
    fn json_to_decimal_array(
        values: &[&Value],
        precision: u8,
        scale: i8,
    ) -> Result<Decimal128Array, ColumnConversionError> {
        let mut builder = Decimal128Builder::new()
            .with_precision_and_scale(precision, scale)
            .map_err(|e| ConversionError::ArrowError(e.to_string()))?;

        for (row, value) in values.iter().enumerate() {
            let scaled = Self::scale_to_decimal(value, scale).map_err(|failure| {
                ColumnConversionError::Value {
                    row,
                    reason: failure.reason,
                    overflow: failure.overflow,
                }
            })?;
            builder.append_option(scaled);
        }

        Ok(builder.finish())
    }

    /// Scale a single JSON value into the i128 representation of a DECIMAL column.
    ///
    /// Exasol sends decimals as strings when they exceed the JSON number range
    /// and as numbers otherwise, so all three shapes are accepted; a value of
    /// any other shape becomes NULL.
    fn scale_to_decimal(value: &Value, scale: i8) -> Result<Option<i128>, DecimalParseFailure> {
        if let Some(text) = value.as_str() {
            return Self::parse_string_to_decimal(text, scale).map(Some);
        }
        if let Some(integer) = value.as_i64() {
            return Ok(Some((integer * 10i64.pow(scale as u32)) as i128));
        }
        if let Some(float) = value.as_f64() {
            return Ok(Some((float * 10f64.powi(scale as i32)) as i128));
        }
        Ok(None)
    }

    /// Parse a decimal string to i128 scaled value.
    ///
    /// Handles formats like "123", "123.45", "-123.45"
    fn parse_string_to_decimal(s: &str, scale: i8) -> Result<i128, DecimalParseFailure> {
        if s.is_empty() {
            return Err(DecimalParseFailure::invalid("Empty decimal string"));
        }

        let parts: Vec<&str> = s.split('.').collect();

        let (integer_part, decimal_part) = match parts.len() {
            1 => (parts[0], ""),
            2 => (parts[0], parts[1]),
            _ => {
                return Err(DecimalParseFailure::invalid(format!(
                    "Invalid decimal format: {}",
                    s
                )));
            }
        };

        let mut result: i128 = integer_part
            .parse()
            .map_err(|e: ParseIntError| match e.kind() {
                IntErrorKind::PosOverflow | IntErrorKind::NegOverflow => {
                    DecimalParseFailure::overflow(format!(
                        "integer part {integer_part} exceeds the 128-bit range"
                    ))
                }
                _ => DecimalParseFailure::invalid(format!("Invalid integer part: {integer_part}")),
            })?;

        result = result
            .checked_mul(10_i128.pow(scale as u32))
            .ok_or_else(|| DecimalParseFailure::overflow("Decimal overflow"))?;

        if !decimal_part.is_empty() {
            let decimal_digits = decimal_part.len().min(scale as usize);
            let decimal_value: i128 = decimal_part[..decimal_digits].parse().map_err(|_| {
                DecimalParseFailure::invalid(format!("Invalid decimal part: {}", decimal_part))
            })?;

            let scale_diff = scale as usize - decimal_digits;
            let scaled_decimal = decimal_value * 10_i128.pow(scale_diff as u32);

            result = result
                .checked_add(if integer_part.starts_with('-') {
                    -scaled_decimal
                } else {
                    scaled_decimal
                })
                .ok_or_else(|| DecimalParseFailure::overflow("Decimal overflow"))?;
        }

        Ok(result)
    }
}

/// Why a DECIMAL string did not scale to an i128, before the column context is known.
#[derive(Debug)]
struct DecimalParseFailure {
    reason: String,
    overflow: bool,
}

impl DecimalParseFailure {
    fn invalid(reason: impl Into<String>) -> Self {
        Self {
            reason: reason.into(),
            overflow: false,
        }
    }

    fn overflow(reason: impl Into<String>) -> Self {
        Self {
            reason: reason.into(),
            overflow: true,
        }
    }
}

/// Iterator over RecordBatches from a result set.
///
/// Lazily fetches data from the transport as needed.
pub struct ResultSetIterator {
    /// Result set handle
    handle: Option<ResultSetHandle>,
    /// Transport reference
    transport: Arc<Mutex<dyn TransportProtocol>>,
    /// Query metadata
    metadata: QueryMetadata,
    /// Buffered batches
    batches: Vec<RecordBatch>,
    /// Current iteration index
    current_index: usize,
    /// Whether all data has been fetched
    complete: bool,
}

impl ResultSetIterator {
    /// Get the query metadata.
    pub fn metadata(&self) -> &QueryMetadata {
        &self.metadata
    }

    /// Fetch the next batch from the transport.
    async fn fetch_next_batch(&mut self) -> Result<Option<RecordBatch>, QueryError> {
        if self.complete {
            return Ok(None);
        }

        let handle = match self.handle {
            Some(h) => h,
            None => {
                self.complete = true;
                return Ok(None);
            }
        };

        let rows_received = ResultSet::row_count_of(&self.batches);
        match ResultSet::fetch_page_after(&self.transport, handle, &self.metadata, rows_received)
            .await
        {
            Ok(None) => {
                self.complete = true;
                Ok(None)
            }
            Ok(batch) => Ok(batch),
            Err(error @ PageFetchError::RowCountMismatch(_)) => {
                self.complete = true;
                Err(error.into())
            }
            Err(error) => Err(error.into()),
        }
    }

    /// Get the next batch synchronously (blocking).
    ///
    /// This is useful for implementing sync Iterator trait.
    ///
    /// # Errors
    /// Returns `QueryError::ExecutionFailed` when the result set delivers fewer or
    /// more rows than its total row count. After that error, every later call
    /// returns `None`.
    pub fn next_batch(&mut self) -> Option<Result<RecordBatch, QueryError>> {
        // Return buffered batch if available
        if self.current_index < self.batches.len() {
            let batch = self.batches[self.current_index].clone();
            self.current_index += 1;
            return Some(Ok(batch));
        }

        // Need to fetch more data
        if self.complete {
            return None;
        }

        // Use block_on for async fetch (not ideal, but works for Phase 1)
        let runtime = tokio::runtime::Handle::try_current();
        if let Ok(handle) = runtime {
            let result = handle.block_on(self.fetch_next_batch());
            match result {
                Ok(Some(batch)) => {
                    self.batches.push(batch.clone());
                    self.current_index += 1;
                    Some(Ok(batch))
                }
                Ok(None) => None,
                Err(e) => Some(Err(e)),
            }
        } else {
            // No runtime available
            Some(Err(QueryError::InvalidState(
                "No async runtime available".to_string(),
            )))
        }
    }

    /// Close the result set and release resources.
    pub async fn close(mut self) -> Result<(), QueryError> {
        if let Some(handle) = self.handle.take() {
            let mut transport = self.transport.lock().await;
            transport
                .close_result_set(handle)
                .await
                .map_err(|e| QueryError::ExecutionFailed(e.to_string()))?;
        }
        Ok(())
    }
}

// Implement Iterator for ResultSetIterator
impl Iterator for ResultSetIterator {
    type Item = Result<RecordBatch, QueryError>;

    fn next(&mut self) -> Option<Self::Item> {
        self.next_batch()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::error::TransportError;
    use crate::transport::messages::{ColumnInfo, DataType, ResultPayload};
    use crate::transport::test_support::MockTransport;
    use crate::types::TypeMapper;
    use arrow::array::Array;
    use mockall::predicate::eq;
    use serde_json::json;

    // =========================================================================
    // Shared test fixtures
    // =========================================================================

    /// An Exasol type descriptor carrying no modifiers.
    fn plain_type(type_name: &str) -> DataType {
        DataType {
            type_name: type_name.to_string(),
            precision: None,
            scale: None,
            size: None,
            character_set: None,
            with_local_time_zone: None,
            fraction: None,
            srid: None,
        }
    }

    fn arrow_type_of(data_type: &DataType) -> Result<arrow::datatypes::DataType, ConversionError> {
        TypeMapper::exasol_to_arrow(&exasol_type_of(data_type)?, true)
    }

    fn decimal_column(name: &str) -> ColumnInfo {
        ColumnInfo {
            name: name.to_string(),
            data_type: DataType {
                precision: Some(18),
                scale: Some(0),
                ..plain_type("DECIMAL")
            },
        }
    }

    fn varchar_column(name: &str) -> ColumnInfo {
        ColumnInfo {
            name: name.to_string(),
            data_type: DataType {
                size: Some(100),
                character_set: Some("UTF8".to_string()),
                ..plain_type("VARCHAR")
            },
        }
    }

    fn boolean_column(name: &str) -> ColumnInfo {
        ColumnInfo {
            name: name.to_string(),
            data_type: plain_type("BOOLEAN"),
        }
    }

    /// A transport that serves one further page holding `values`, then reports
    /// exhaustion with an empty page.
    fn transport_serving_one_more_page(values: &'static [i64]) -> MockTransport {
        let mut transport = MockTransport::new();
        let mut call = 0;
        transport
            .expect_fetch_results()
            .times(2)
            .returning(move |_| {
                call += 1;
                Ok(if call == 1 {
                    single_column_result_data(values, 0)
                } else {
                    single_column_result_data(&[], 0)
                })
            });
        transport
    }

    fn single_column_result_data(values: &[i64], total_rows: i64) -> ResultData {
        ResultData {
            columns: vec![decimal_column("id")],
            data: ResultPayload::Json(values.iter().map(|v| vec![json!(v)]).collect()),
            total_rows,
        }
    }

    fn streaming_result_set(
        transport: MockTransport,
        values: &[i64],
        total_rows: i64,
        handle: Option<ResultSetHandle>,
    ) -> ResultSet {
        let transport: Arc<Mutex<dyn TransportProtocol>> = Arc::new(Mutex::new(transport));
        ResultSet::from_transport_result(
            TransportQueryResult::ResultSet {
                handle,
                data: single_column_result_data(values, total_rows),
            },
            transport,
        )
        .unwrap()
    }

    fn streaming_iterator(
        transport: MockTransport,
        values: &[i64],
        total_rows: i64,
        handle: Option<ResultSetHandle>,
    ) -> ResultSetIterator {
        streaming_result_set(transport, values, total_rows, handle)
            .into_iterator()
            .unwrap()
    }

    fn entered_runtime() -> tokio::runtime::Runtime {
        tokio::runtime::Builder::new_current_thread()
            .build()
            .expect("current-thread runtime")
    }

    #[tokio::test]
    async fn test_result_set_row_count() {
        let mock_transport = MockTransport::new();
        let transport: Arc<Mutex<dyn TransportProtocol>> = Arc::new(Mutex::new(mock_transport));

        let result = TransportQueryResult::RowCount { count: 42 };
        let result_set = ResultSet::from_transport_result(result, transport).unwrap();

        assert_eq!(result_set.row_count(), Some(42));
        assert!(!result_set.is_stream());
        assert!(result_set.metadata().is_none());
    }

    #[tokio::test]
    async fn test_result_set_stream() {
        let mock_transport = MockTransport::new();
        let transport: Arc<Mutex<dyn TransportProtocol>> = Arc::new(Mutex::new(mock_transport));

        // Row-major data format:
        // Row 0: [1, "Alice"]
        // Row 1: [2, "Bob"]
        let data = ResultData {
            columns: vec![decimal_column("id"), varchar_column("name")],
            data: ResultPayload::Json(vec![
                vec![serde_json::json!(1), serde_json::json!("Alice")],
                vec![serde_json::json!(2), serde_json::json!("Bob")],
            ]),
            total_rows: 2,
        };

        let result = TransportQueryResult::ResultSet {
            handle: Some(ResultSetHandle::new(1)),
            data,
        };

        let result_set = ResultSet::from_transport_result(result, transport).unwrap();

        assert!(result_set.row_count().is_none());
        assert!(result_set.is_stream());

        let metadata = result_set.metadata().unwrap();
        assert_eq!(metadata.column_count, 2);
        assert_eq!(metadata.total_rows, Some(2));
        assert_eq!(metadata.column_names(), vec!["id", "name"]);
    }

    #[tokio::test]
    async fn test_empty_result_set_preserves_schema() {
        // A zero-row result set must still convey its columns: ADBC consumers
        // read the schema from the first batch, so an empty Vec would drop it
        // (regression test for `WHERE FALSE LIMIT 0` schema probes).
        let mock_transport = MockTransport::new();
        let transport: Arc<Mutex<dyn TransportProtocol>> = Arc::new(Mutex::new(mock_transport));

        let data = ResultData {
            columns: vec![decimal_column("id"), varchar_column("name")],
            data: ResultPayload::Json(vec![]),
            total_rows: 0,
        };

        let result = TransportQueryResult::ResultSet { handle: None, data };
        let result_set = ResultSet::from_transport_result(result, transport).unwrap();

        let batches = result_set.fetch_all().await.unwrap();
        assert_eq!(
            batches.len(),
            1,
            "empty result set must yield one schema-carrying batch"
        );
        assert_eq!(batches[0].num_rows(), 0);
        assert_eq!(batches[0].num_columns(), 2);
        assert_eq!(
            batches[0]
                .schema()
                .fields()
                .iter()
                .map(|f| f.name().clone())
                .collect::<Vec<_>>(),
            vec!["id", "name"]
        );
    }

    #[tokio::test]
    async fn test_result_set_to_record_batch() {
        let schema = Arc::new(Schema::new(vec![
            Field::new("id", arrow::datatypes::DataType::Decimal128(18, 0), true),
            Field::new("name", arrow::datatypes::DataType::Utf8, true),
            Field::new("active", arrow::datatypes::DataType::Boolean, true),
        ]));

        // Row-major data format:
        // Row 0: [1, "Alice", true]
        // Row 1: [2, "Bob", false]
        let rows = vec![
            vec![
                serde_json::json!(1),
                serde_json::json!("Alice"),
                serde_json::json!(true),
            ],
            vec![
                serde_json::json!(2),
                serde_json::json!("Bob"),
                serde_json::json!(false),
            ],
        ];

        let batch = ResultSet::column_major_to_record_batch(&rows, &schema).unwrap();

        assert_eq!(batch.num_rows(), 2);
        assert_eq!(batch.num_columns(), 3);
    }

    #[tokio::test]
    async fn test_single_row_single_column() {
        // Test case for: SELECT 42 AS answer
        let schema = Arc::new(Schema::new(vec![Field::new(
            "answer",
            arrow::datatypes::DataType::Decimal128(18, 0),
            true,
        )]));

        // Row-major: one row with one value
        let rows = vec![
            vec![serde_json::json!(42)], // Row 0: [42]
        ];

        let batch = ResultSet::column_major_to_record_batch(&rows, &schema).unwrap();

        assert_eq!(batch.num_rows(), 1, "Should have exactly 1 row");
        assert_eq!(batch.num_columns(), 1, "Should have exactly 1 column");
    }

    #[tokio::test]
    async fn test_single_row_two_columns() {
        // Test case for: SELECT 42 AS answer, 'hello' AS greeting
        let schema = Arc::new(Schema::new(vec![
            Field::new(
                "answer",
                arrow::datatypes::DataType::Decimal128(18, 0),
                true,
            ),
            Field::new("greeting", arrow::datatypes::DataType::Utf8, true),
        ]));

        // Row-major: one row with two values
        let rows = vec![
            vec![serde_json::json!(42), serde_json::json!("hello")], // Row 0: [42, "hello"]
        ];

        let batch = ResultSet::column_major_to_record_batch(&rows, &schema).unwrap();

        assert_eq!(batch.num_rows(), 1, "Should have exactly 1 row");
        assert_eq!(batch.num_columns(), 2, "Should have exactly 2 columns");
    }

    #[tokio::test]
    async fn test_ten_rows_two_columns() {
        // Test case for: SELECT LEVEL AS id, 'Row ' || LEVEL AS label FROM DUAL CONNECT BY LEVEL <= 10
        let schema = Arc::new(Schema::new(vec![
            Field::new("id", arrow::datatypes::DataType::Decimal128(18, 0), true),
            Field::new("label", arrow::datatypes::DataType::Utf8, true),
        ]));

        // Row-major: ten rows, each with two values
        let rows = vec![
            vec![serde_json::json!(1), serde_json::json!("Row 1")],
            vec![serde_json::json!(2), serde_json::json!("Row 2")],
            vec![serde_json::json!(3), serde_json::json!("Row 3")],
            vec![serde_json::json!(4), serde_json::json!("Row 4")],
            vec![serde_json::json!(5), serde_json::json!("Row 5")],
            vec![serde_json::json!(6), serde_json::json!("Row 6")],
            vec![serde_json::json!(7), serde_json::json!("Row 7")],
            vec![serde_json::json!(8), serde_json::json!("Row 8")],
            vec![serde_json::json!(9), serde_json::json!("Row 9")],
            vec![serde_json::json!(10), serde_json::json!("Row 10")],
        ];

        let batch = ResultSet::column_major_to_record_batch(&rows, &schema).unwrap();

        assert_eq!(batch.num_rows(), 10, "Should have exactly 10 rows");
        assert_eq!(batch.num_columns(), 2, "Should have exactly 2 columns");
    }

    #[tokio::test]
    async fn test_schema_building() {
        let columns = vec![
            decimal_column("id"),
            varchar_column("name"),
            ColumnInfo {
                name: "created_at".to_string(),
                data_type: DataType {
                    with_local_time_zone: Some(false),
                    ..plain_type("TIMESTAMP")
                },
            },
        ];

        let schema = ResultSet::build_schema(&columns).unwrap();

        assert_eq!(schema.fields().len(), 3);
        assert_eq!(schema.field(0).name(), "id");
        assert_eq!(schema.field(1).name(), "name");
        assert_eq!(schema.field(2).name(), "created_at");
    }

    #[test]
    fn test_query_metadata() {
        let schema = Arc::new(Schema::new(vec![
            Field::new("id", arrow::datatypes::DataType::Int64, false),
            Field::new("name", arrow::datatypes::DataType::Utf8, true),
        ]));

        let metadata = QueryMetadata::new(Arc::clone(&schema), Some(100)).with_execution_time(250);

        assert_eq!(metadata.column_count, 2);
        assert_eq!(metadata.total_rows, Some(100));
        assert_eq!(metadata.execution_time_ms, Some(250));
        assert_eq!(metadata.column_names(), vec!["id", "name"]);
    }

    // =========================================================================
    // Tests for QueryMetadata::column_types
    // =========================================================================

    #[test]
    fn test_query_metadata_column_types_returns_correct_types() {
        let schema = Arc::new(Schema::new(vec![
            Field::new("id", arrow::datatypes::DataType::Int64, false),
            Field::new("name", arrow::datatypes::DataType::Utf8, true),
            Field::new("active", arrow::datatypes::DataType::Boolean, true),
        ]));

        let metadata = QueryMetadata::new(Arc::clone(&schema), Some(100));
        let types = metadata.column_types();

        assert_eq!(types.len(), 3);
        assert_eq!(types[0], &arrow::datatypes::DataType::Int64);
        assert_eq!(types[1], &arrow::datatypes::DataType::Utf8);
        assert_eq!(types[2], &arrow::datatypes::DataType::Boolean);
    }

    #[test]
    fn test_query_metadata_column_types_empty_schema() {
        let fields: Vec<Field> = vec![];
        let schema = Arc::new(Schema::new(fields));

        let metadata = QueryMetadata::new(Arc::clone(&schema), None);
        let types = metadata.column_types();

        assert!(types.is_empty());
    }

    // =========================================================================
    // Tests for ResultSet::parse_string_to_decimal
    // =========================================================================

    #[test]
    fn test_parse_string_to_decimal_integer() {
        let result = ResultSet::parse_string_to_decimal("123", 2).unwrap();
        // 123 with scale 2 = 12300
        assert_eq!(result, 12300);
    }

    #[test]
    fn test_parse_string_to_decimal_with_decimal_point() {
        let result = ResultSet::parse_string_to_decimal("123.45", 2).unwrap();
        // 123.45 with scale 2 = 12345
        assert_eq!(result, 12345);
    }

    #[test]
    fn test_parse_string_to_decimal_negative() {
        let result = ResultSet::parse_string_to_decimal("-123.45", 2).unwrap();
        // -123.45 with scale 2 = -12345
        assert_eq!(result, -12345);
    }

    #[test]
    fn test_parse_string_to_decimal_zero_scale() {
        let result = ResultSet::parse_string_to_decimal("123", 0).unwrap();
        assert_eq!(result, 123);
    }

    #[test]
    fn test_parse_string_to_decimal_high_scale() {
        let result = ResultSet::parse_string_to_decimal("1.5", 6).unwrap();
        // 1.5 with scale 6 = 1500000
        assert_eq!(result, 1500000);
    }

    #[test]
    fn test_parse_string_to_decimal_truncates_extra_decimals() {
        // When decimal part has more digits than scale
        let result = ResultSet::parse_string_to_decimal("1.123456", 3).unwrap();
        // 1.123 with scale 3 = 1123 (truncates extra digits)
        assert_eq!(result, 1123);
    }

    #[test]
    fn test_parse_string_to_decimal_zero() {
        let result = ResultSet::parse_string_to_decimal("0", 2).unwrap();
        assert_eq!(result, 0);
    }

    #[test]
    fn test_parse_string_to_decimal_negative_zero() {
        let result = ResultSet::parse_string_to_decimal("-0", 2).unwrap();
        assert_eq!(result, 0);
    }

    #[test]
    fn test_parse_string_to_decimal_empty_string() {
        let result = ResultSet::parse_string_to_decimal("", 2);
        assert!(result.is_err());
    }

    #[test]
    fn test_parse_string_to_decimal_multiple_decimal_points() {
        let result = ResultSet::parse_string_to_decimal("1.2.3", 2);
        assert!(result.is_err());
    }

    #[test]
    fn test_parse_string_to_decimal_non_numeric() {
        let result = ResultSet::parse_string_to_decimal("abc", 2);
        assert!(result.is_err());
    }

    #[test]
    fn test_parse_string_to_decimal_large_value() {
        let result = ResultSet::parse_string_to_decimal("999999999999999999", 0).unwrap();
        assert_eq!(result, 999999999999999999_i128);
    }

    // =========================================================================
    // Tests for exasol_type_of
    // =========================================================================

    #[test]
    fn test_exasol_type_of_boolean() {
        let data_type = plain_type("BOOLEAN");

        let result = arrow_type_of(&data_type).unwrap();
        assert_eq!(result, arrow::datatypes::DataType::Boolean);
    }

    #[test]
    fn test_exasol_type_of_char() {
        let data_type = DataType {
            size: Some(10),
            ..plain_type("CHAR")
        };

        let result = arrow_type_of(&data_type).unwrap();
        assert_eq!(result, arrow::datatypes::DataType::Utf8);
    }

    #[test]
    fn test_exasol_type_of_char_default_size() {
        let data_type = plain_type("CHAR");

        let result = arrow_type_of(&data_type).unwrap();
        assert_eq!(result, arrow::datatypes::DataType::Utf8);
    }

    #[test]
    fn test_exasol_type_of_varchar() {
        let data_type = DataType {
            size: Some(100),
            ..plain_type("VARCHAR")
        };

        let result = arrow_type_of(&data_type).unwrap();
        assert_eq!(result, arrow::datatypes::DataType::Utf8);
    }

    #[test]
    fn test_exasol_type_of_varchar_default_size() {
        let data_type = plain_type("VARCHAR");

        let result = arrow_type_of(&data_type).unwrap();
        assert_eq!(result, arrow::datatypes::DataType::Utf8);
    }

    #[test]
    fn test_exasol_type_of_decimal() {
        let data_type = DataType {
            precision: Some(18),
            scale: Some(2),
            ..plain_type("DECIMAL")
        };

        let result = arrow_type_of(&data_type).unwrap();
        assert_eq!(result, arrow::datatypes::DataType::Decimal128(18, 2));
    }

    #[test]
    fn test_exasol_type_of_decimal_default_precision_scale() {
        let data_type = plain_type("DECIMAL");

        let result = arrow_type_of(&data_type).unwrap();
        assert_eq!(result, arrow::datatypes::DataType::Decimal128(18, 0));
    }

    #[test]
    fn test_exasol_type_of_double() {
        let data_type = plain_type("DOUBLE");

        let result = arrow_type_of(&data_type).unwrap();
        assert_eq!(result, arrow::datatypes::DataType::Float64);
    }

    #[test]
    fn test_exasol_type_of_date() {
        let data_type = plain_type("DATE");

        let result = arrow_type_of(&data_type).unwrap();
        assert_eq!(result, arrow::datatypes::DataType::Date32);
    }

    #[test]
    fn test_exasol_type_of_timestamp_without_tz() {
        let data_type = DataType {
            with_local_time_zone: Some(false),
            ..plain_type("TIMESTAMP")
        };

        let result = arrow_type_of(&data_type).unwrap();
        assert!(matches!(
            result,
            arrow::datatypes::DataType::Timestamp(arrow::datatypes::TimeUnit::Microsecond, None)
        ));
    }

    #[test]
    fn test_exasol_type_of_timestamp_with_tz() {
        let data_type = DataType {
            with_local_time_zone: Some(true),
            ..plain_type("TIMESTAMP")
        };

        let result = arrow_type_of(&data_type).unwrap();
        assert!(matches!(
            result,
            arrow::datatypes::DataType::Timestamp(arrow::datatypes::TimeUnit::Microsecond, Some(_))
        ));
    }

    #[test]
    fn test_exasol_type_of_timestamp_default_tz() {
        let data_type = plain_type("TIMESTAMP");

        let result = arrow_type_of(&data_type).unwrap();
        assert!(matches!(
            result,
            arrow::datatypes::DataType::Timestamp(arrow::datatypes::TimeUnit::Microsecond, None)
        ));
    }

    #[test]
    fn test_exasol_type_of_interval_year_to_month() {
        let data_type = plain_type("INTERVAL YEAR TO MONTH");

        let result = arrow_type_of(&data_type).unwrap();
        assert!(matches!(
            result,
            arrow::datatypes::DataType::Interval(arrow::datatypes::IntervalUnit::MonthDayNano)
        ));
    }

    #[test]
    fn test_exasol_type_of_interval_day_to_second() {
        let data_type = DataType {
            fraction: Some(6),
            ..plain_type("INTERVAL DAY TO SECOND")
        };

        let result = arrow_type_of(&data_type).unwrap();
        assert!(matches!(
            result,
            arrow::datatypes::DataType::Interval(arrow::datatypes::IntervalUnit::MonthDayNano)
        ));
    }

    #[test]
    fn test_exasol_type_of_interval_day_to_second_default_fraction() {
        let data_type = plain_type("INTERVAL DAY TO SECOND");

        let result = arrow_type_of(&data_type).unwrap();
        assert!(matches!(
            result,
            arrow::datatypes::DataType::Interval(arrow::datatypes::IntervalUnit::MonthDayNano)
        ));
    }

    #[test]
    fn test_exasol_type_of_geometry() {
        let data_type = plain_type("GEOMETRY");

        let result = arrow_type_of(&data_type).unwrap();
        assert_eq!(result, arrow::datatypes::DataType::Binary);
    }

    #[test]
    fn test_exasol_type_of_hashtype() {
        let data_type = plain_type("HASHTYPE");

        let result = arrow_type_of(&data_type).unwrap();
        assert_eq!(result, arrow::datatypes::DataType::Binary);
        assert_eq!(
            exasol_type_of(&data_type).unwrap(),
            ExasolType::Hashtype { byte_size: 16 }
        );
    }

    #[test]
    fn test_exasol_type_of_hashtype_uses_reported_size() {
        let data_type = DataType {
            size: Some(64),
            ..plain_type("HASHTYPE")
        };

        assert_eq!(
            exasol_type_of(&data_type).unwrap(),
            ExasolType::Hashtype { byte_size: 32 }
        );
    }

    #[test]
    fn test_exasol_type_of_unsupported_type() {
        let data_type = plain_type("UNKNOWN_TYPE");

        let result = arrow_type_of(&data_type);
        assert!(result.is_err());
    }

    // =========================================================================
    // Tests for ResultSet::column_major_to_record_batch with various types
    // =========================================================================

    #[tokio::test]
    async fn test_column_major_to_record_batch_with_int32() {
        let schema = Arc::new(Schema::new(vec![Field::new(
            "value",
            arrow::datatypes::DataType::Int32,
            true,
        )]));

        let rows = vec![
            vec![serde_json::json!(1)],
            vec![serde_json::json!(2)],
            vec![serde_json::json!(null)],
            vec![serde_json::json!(4)],
        ];

        let batch = ResultSet::column_major_to_record_batch(&rows, &schema).unwrap();

        assert_eq!(batch.num_rows(), 4);
        assert_eq!(batch.num_columns(), 1);

        let array = batch
            .column(0)
            .as_any()
            .downcast_ref::<arrow::array::Int32Array>()
            .unwrap();
        assert_eq!(array.value(0), 1);
        assert_eq!(array.value(1), 2);
        assert!(array.is_null(2));
        assert_eq!(array.value(3), 4);
    }

    #[tokio::test]
    async fn test_column_major_to_record_batch_with_int64() {
        let schema = Arc::new(Schema::new(vec![Field::new(
            "value",
            arrow::datatypes::DataType::Int64,
            true,
        )]));

        let rows = vec![
            vec![serde_json::json!(9223372036854775807_i64)], // Max i64
            vec![serde_json::json!(-9223372036854775808_i64)], // Min i64
            vec![serde_json::json!(null)],
        ];

        let batch = ResultSet::column_major_to_record_batch(&rows, &schema).unwrap();

        assert_eq!(batch.num_rows(), 3);

        let array = batch
            .column(0)
            .as_any()
            .downcast_ref::<arrow::array::Int64Array>()
            .unwrap();
        assert_eq!(array.value(0), 9223372036854775807_i64);
        assert_eq!(array.value(1), -9223372036854775808_i64);
        assert!(array.is_null(2));
    }

    #[tokio::test]
    async fn test_column_major_to_record_batch_with_float64() {
        let schema = Arc::new(Schema::new(vec![Field::new(
            "value",
            arrow::datatypes::DataType::Float64,
            true,
        )]));

        let rows = vec![
            vec![serde_json::json!(1.23456)],
            vec![serde_json::json!(-9.87654)],
            vec![serde_json::json!(null)],
            vec![serde_json::json!(0.0)],
        ];

        let batch = ResultSet::column_major_to_record_batch(&rows, &schema).unwrap();

        assert_eq!(batch.num_rows(), 4);

        let array = batch
            .column(0)
            .as_any()
            .downcast_ref::<arrow::array::Float64Array>()
            .unwrap();
        assert!((array.value(0) - 1.23456).abs() < 0.00001);
        assert!((array.value(1) - (-9.87654)).abs() < 0.00001);
        assert!(array.is_null(2));
        assert!((array.value(3) - 0.0).abs() < 0.00001);
    }

    #[tokio::test]
    async fn test_column_major_to_record_batch_with_boolean() {
        let schema = Arc::new(Schema::new(vec![Field::new(
            "flag",
            arrow::datatypes::DataType::Boolean,
            true,
        )]));

        let rows = vec![
            vec![serde_json::json!(true)],
            vec![serde_json::json!(false)],
            vec![serde_json::json!(null)],
        ];

        let batch = ResultSet::column_major_to_record_batch(&rows, &schema).unwrap();

        assert_eq!(batch.num_rows(), 3);

        let array = batch
            .column(0)
            .as_any()
            .downcast_ref::<arrow::array::BooleanArray>()
            .unwrap();
        assert!(array.value(0));
        assert!(!array.value(1));
        assert!(array.is_null(2));
    }

    #[tokio::test]
    async fn test_column_major_to_record_batch_with_date32() {
        let schema = Arc::new(Schema::new(vec![Field::new(
            "date",
            arrow::datatypes::DataType::Date32,
            true,
        )]));

        let rows = vec![
            vec![serde_json::json!("1970-01-01")],
            vec![serde_json::json!("2000-01-01")],
            vec![serde_json::json!(null)],
        ];

        let batch = ResultSet::column_major_to_record_batch(&rows, &schema).unwrap();

        assert_eq!(batch.num_rows(), 3);

        let array = batch
            .column(0)
            .as_any()
            .downcast_ref::<arrow::array::Date32Array>()
            .unwrap();
        assert_eq!(array.value(0), 0); // Unix epoch
        assert_eq!(array.value(1), 10957); // 2000-01-01
        assert!(array.is_null(2));
    }

    /// Scenario: Pre-1970 DATE and TIMESTAMP query results keep their calendar day
    #[tokio::test]
    async fn test_column_major_to_record_batch_keeps_pre_1970_dates_and_timestamps() {
        let schema = Arc::new(Schema::new(vec![
            Field::new("date", arrow::datatypes::DataType::Date32, true),
            Field::new(
                "timestamp",
                arrow::datatypes::DataType::Timestamp(
                    arrow::datatypes::TimeUnit::Microsecond,
                    None,
                ),
                true,
            ),
        ]));

        let rows = vec![
            vec![
                serde_json::json!("1968-01-01"),
                serde_json::json!("1950-06-15 00:00:00"),
            ],
            vec![
                serde_json::json!("0001-01-01"),
                serde_json::json!("1969-12-31 23:59:59.999999"),
            ],
        ];

        let batch = ResultSet::column_major_to_record_batch(&rows, &schema).unwrap();

        let dates = batch
            .column(0)
            .as_any()
            .downcast_ref::<arrow::array::Date32Array>()
            .unwrap();
        assert_eq!(dates.value(0), -731);
        assert_eq!(dates.value(1), -719162);
        let timestamps = batch
            .column(1)
            .as_any()
            .downcast_ref::<arrow::array::TimestampMicrosecondArray>()
            .unwrap();
        assert_eq!(timestamps.value(0), -616_896_000_000_000);
        assert_eq!(timestamps.value(1), -1);
    }

    #[tokio::test]
    async fn test_column_major_to_record_batch_with_timestamp() {
        let schema = Arc::new(Schema::new(vec![Field::new(
            "timestamp",
            arrow::datatypes::DataType::Timestamp(arrow::datatypes::TimeUnit::Microsecond, None),
            true,
        )]));

        let rows = vec![
            vec![serde_json::json!("1970-01-01 00:00:00")],
            vec![serde_json::json!("1970-01-01 01:00:00.123456")],
            vec![serde_json::json!(null)],
        ];

        let batch = ResultSet::column_major_to_record_batch(&rows, &schema).unwrap();

        assert_eq!(batch.num_rows(), 3);

        let array = batch
            .column(0)
            .as_any()
            .downcast_ref::<arrow::array::TimestampMicrosecondArray>()
            .unwrap();
        assert_eq!(array.value(0), 0);
        // 1 hour + 123456 microseconds
        assert_eq!(array.value(1), 3600 * 1_000_000 + 123456);
        assert!(array.is_null(2));
    }

    #[tokio::test]
    async fn test_column_major_to_record_batch_empty_data() {
        let schema = Arc::new(Schema::new(vec![
            Field::new("id", arrow::datatypes::DataType::Int64, true),
            Field::new("name", arrow::datatypes::DataType::Utf8, true),
        ]));

        let rows = vec![];

        let batch = ResultSet::column_major_to_record_batch(&rows, &schema).unwrap();

        assert_eq!(batch.num_rows(), 0);
        assert_eq!(batch.num_columns(), 2);
    }

    #[tokio::test]
    async fn test_column_major_to_record_batch_with_utf8_string() {
        let schema = Arc::new(Schema::new(vec![Field::new(
            "text",
            arrow::datatypes::DataType::Utf8,
            true,
        )]));

        let rows = vec![
            vec![serde_json::json!("hello")],
            vec![serde_json::json!("world")],
            vec![serde_json::json!(null)],
            vec![serde_json::json!("")], // Empty string
        ];

        let batch = ResultSet::column_major_to_record_batch(&rows, &schema).unwrap();

        assert_eq!(batch.num_rows(), 4);

        let array = batch
            .column(0)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(array.value(0), "hello");
        assert_eq!(array.value(1), "world");
        assert!(array.is_null(2));
        assert_eq!(array.value(3), "");
    }

    #[tokio::test]
    async fn test_column_major_to_record_batch_utf8_from_non_string_value() {
        // Test that non-string JSON values are converted to string
        let schema = Arc::new(Schema::new(vec![Field::new(
            "text",
            arrow::datatypes::DataType::Utf8,
            true,
        )]));

        let rows = vec![
            vec![serde_json::json!(123)], // Number should be converted to "123"
        ];

        let batch = ResultSet::column_major_to_record_batch(&rows, &schema).unwrap();

        let array = batch
            .column(0)
            .as_any()
            .downcast_ref::<arrow::array::StringArray>()
            .unwrap();
        assert_eq!(array.value(0), "123");
    }

    #[tokio::test]
    async fn test_column_major_to_record_batch_decimal_from_string() {
        let schema = Arc::new(Schema::new(vec![Field::new(
            "amount",
            arrow::datatypes::DataType::Decimal128(18, 2),
            true,
        )]));

        let rows = vec![
            vec![serde_json::json!("123.45")],
            vec![serde_json::json!("-67.89")],
            vec![serde_json::json!(null)],
        ];

        let batch = ResultSet::column_major_to_record_batch(&rows, &schema).unwrap();

        assert_eq!(batch.num_rows(), 3);

        let array = batch
            .column(0)
            .as_any()
            .downcast_ref::<arrow::array::Decimal128Array>()
            .unwrap();
        assert_eq!(array.value(0), 12345); // 123.45 with scale 2
        assert_eq!(array.value(1), -6789); // -67.89 with scale 2
        assert!(array.is_null(2));
    }

    #[tokio::test]
    async fn test_column_major_to_record_batch_decimal_from_integer() {
        let schema = Arc::new(Schema::new(vec![Field::new(
            "amount",
            arrow::datatypes::DataType::Decimal128(18, 2),
            true,
        )]));

        let rows = vec![
            vec![serde_json::json!(100)], // Integer should be scaled
        ];

        let batch = ResultSet::column_major_to_record_batch(&rows, &schema).unwrap();

        let array = batch
            .column(0)
            .as_any()
            .downcast_ref::<arrow::array::Decimal128Array>()
            .unwrap();
        assert_eq!(array.value(0), 10000); // 100 with scale 2 = 10000
    }

    #[tokio::test]
    async fn test_column_major_to_record_batch_decimal_from_float() {
        let schema = Arc::new(Schema::new(vec![Field::new(
            "amount",
            arrow::datatypes::DataType::Decimal128(18, 2),
            true,
        )]));

        let rows = vec![
            vec![serde_json::json!(99.99)], // Float should be scaled
        ];

        let batch = ResultSet::column_major_to_record_batch(&rows, &schema).unwrap();

        let array = batch
            .column(0)
            .as_any()
            .downcast_ref::<arrow::array::Decimal128Array>()
            .unwrap();
        assert_eq!(array.value(0), 9999); // 99.99 with scale 2 = 9999
    }

    #[tokio::test]
    async fn test_column_major_to_record_batch_invalid_date_becomes_null() {
        let schema = Arc::new(Schema::new(vec![Field::new(
            "date",
            arrow::datatypes::DataType::Date32,
            true,
        )]));

        let rows = vec![
            vec![serde_json::json!("invalid-date")],
            vec![serde_json::json!(123)], // Non-string should become null
        ];

        let batch = ResultSet::column_major_to_record_batch(&rows, &schema).unwrap();

        let array = batch
            .column(0)
            .as_any()
            .downcast_ref::<arrow::array::Date32Array>()
            .unwrap();
        // Invalid date formats become null
        assert!(array.is_null(0));
        assert!(array.is_null(1));
    }

    #[tokio::test]
    async fn test_column_major_to_record_batch_invalid_timestamp_becomes_null() {
        let schema = Arc::new(Schema::new(vec![Field::new(
            "timestamp",
            arrow::datatypes::DataType::Timestamp(arrow::datatypes::TimeUnit::Microsecond, None),
            true,
        )]));

        let rows = vec![
            vec![serde_json::json!("not-a-timestamp")],
            vec![serde_json::json!(12345)], // Non-string should become null
        ];

        let batch = ResultSet::column_major_to_record_batch(&rows, &schema).unwrap();

        let array = batch
            .column(0)
            .as_any()
            .downcast_ref::<arrow::array::TimestampMicrosecondArray>()
            .unwrap();
        assert!(array.is_null(0));
        assert!(array.is_null(1));
    }

    #[tokio::test]
    async fn test_column_major_to_record_batch_boolean_null_from_non_bool() {
        let schema = Arc::new(Schema::new(vec![Field::new(
            "flag",
            arrow::datatypes::DataType::Boolean,
            true,
        )]));

        let rows = vec![
            vec![serde_json::json!("not-a-bool")], // String is not a bool
            vec![serde_json::json!(123)],          // Number is not a bool
        ];

        let batch = ResultSet::column_major_to_record_batch(&rows, &schema).unwrap();

        let array = batch
            .column(0)
            .as_any()
            .downcast_ref::<arrow::array::BooleanArray>()
            .unwrap();
        // Non-bool values become null
        assert!(array.is_null(0));
        assert!(array.is_null(1));
    }

    #[tokio::test]
    async fn test_column_major_to_record_batch_int32_null_from_non_int() {
        let schema = Arc::new(Schema::new(vec![Field::new(
            "value",
            arrow::datatypes::DataType::Int32,
            true,
        )]));

        let rows = vec![
            vec![serde_json::json!("not-an-int")], // String is not parseable as int
        ];

        let batch = ResultSet::column_major_to_record_batch(&rows, &schema).unwrap();

        let array = batch
            .column(0)
            .as_any()
            .downcast_ref::<arrow::array::Int32Array>()
            .unwrap();
        assert!(array.is_null(0));
    }

    #[tokio::test]
    async fn test_column_major_to_record_batch_float64_null_from_non_float() {
        let schema = Arc::new(Schema::new(vec![Field::new(
            "value",
            arrow::datatypes::DataType::Float64,
            true,
        )]));

        let rows = vec![
            vec![serde_json::json!("not-a-float")], // String is not parseable as float
        ];

        let batch = ResultSet::column_major_to_record_batch(&rows, &schema).unwrap();

        let array = batch
            .column(0)
            .as_any()
            .downcast_ref::<arrow::array::Float64Array>()
            .unwrap();
        assert!(array.is_null(0));
    }

    #[tokio::test]
    async fn test_column_major_to_record_batch_unsupported_type_returns_error() {
        let schema = Arc::new(Schema::new(vec![Field::new(
            "large_text",
            arrow::datatypes::DataType::LargeUtf8,
            true,
        )]));

        let rows = vec![vec![serde_json::json!("test_data")]];

        let result = ResultSet::column_major_to_record_batch(&rows, &schema);

        let message = result.unwrap_err().to_string();
        assert!(message.contains("LargeUtf8"));
        assert!(message.contains("large_text"));
    }

    // =========================================================================
    // Tests for ResultSetIterator::metadata
    // =========================================================================

    #[tokio::test]
    async fn test_result_set_iterator_metadata() {
        let mock_transport = MockTransport::new();
        let transport: Arc<Mutex<dyn TransportProtocol>> = Arc::new(Mutex::new(mock_transport));

        let data = ResultData {
            columns: vec![decimal_column("id")],
            data: ResultPayload::Json(vec![vec![serde_json::json!(1)]]),
            total_rows: 1,
        };

        let result = TransportQueryResult::ResultSet {
            handle: Some(ResultSetHandle::new(1)),
            data,
        };

        let result_set = ResultSet::from_transport_result(result, transport).unwrap();
        let iterator = result_set.into_iterator().unwrap();

        let metadata = iterator.metadata();
        assert_eq!(metadata.column_count, 1);
        assert_eq!(metadata.total_rows, Some(1));
        assert_eq!(metadata.column_names(), vec!["id"]);
    }

    #[tokio::test]
    async fn test_result_set_iterator_metadata_multiple_columns() {
        let mock_transport = MockTransport::new();
        let transport: Arc<Mutex<dyn TransportProtocol>> = Arc::new(Mutex::new(mock_transport));

        let data = ResultData {
            columns: vec![
                decimal_column("id"),
                varchar_column("name"),
                boolean_column("active"),
            ],
            data: ResultPayload::Json(vec![vec![
                serde_json::json!(1),
                serde_json::json!("Alice"),
                serde_json::json!(true),
            ]]),
            total_rows: 1,
        };

        let result = TransportQueryResult::ResultSet {
            handle: Some(ResultSetHandle::new(1)),
            data,
        };

        let result_set = ResultSet::from_transport_result(result, transport).unwrap();
        let iterator = result_set.into_iterator().unwrap();

        let metadata = iterator.metadata();
        assert_eq!(metadata.column_count, 3);
        assert_eq!(metadata.column_names(), vec!["id", "name", "active"]);

        let types = metadata.column_types();
        assert_eq!(types.len(), 3);
        assert!(matches!(
            types[0],
            arrow::datatypes::DataType::Decimal128(18, 0)
        ));
        assert!(matches!(types[1], arrow::datatypes::DataType::Utf8));
        assert!(matches!(types[2], arrow::datatypes::DataType::Boolean));
    }

    // =========================================================================
    // Tests for ResultSet::fetch_all pagination
    // =========================================================================

    #[tokio::test]
    async fn test_fetch_all_paginates_until_empty_payload() {
        let handle = ResultSetHandle::new(7);
        let mut transport = transport_serving_one_more_page(&[2]);
        transport
            .expect_close_result_set()
            .with(eq(handle))
            .times(1)
            .returning(|_| Ok(()));

        let batches = streaming_result_set(transport, &[1], 0, Some(handle))
            .fetch_all()
            .await
            .unwrap();

        assert_eq!(batches.len(), 2);
        assert_eq!(ResultSet::row_count_of(&batches), 2);
    }

    #[tokio::test]
    async fn test_fetch_all_stops_once_known_total_is_reached() {
        let handle = ResultSetHandle::new(1);
        let mut transport = MockTransport::new();
        // Each page reports its own `total_rows`; only the metadata total (3)
        // may end pagination, so exactly one extra fetch must happen.
        transport
            .expect_fetch_results()
            .times(1)
            .returning(|_| Ok(single_column_result_data(&[2, 3], 2)));
        transport
            .expect_close_result_set()
            .times(1)
            .returning(|_| Ok(()));

        let batches = streaming_result_set(transport, &[1], 3, Some(handle))
            .fetch_all()
            .await
            .unwrap();

        assert_eq!(batches.len(), 2);
        assert_eq!(ResultSet::row_count_of(&batches), 3);
    }

    #[tokio::test]
    async fn test_fetch_all_skips_pagination_when_first_page_is_complete() {
        let handle = ResultSetHandle::new(3);
        let mut transport = MockTransport::new();
        transport.expect_fetch_results().times(0);
        transport
            .expect_close_result_set()
            .with(eq(handle))
            .times(1)
            .returning(|_| Ok(()));

        let batches = streaming_result_set(transport, &[1, 2], 2, Some(handle))
            .fetch_all()
            .await
            .unwrap();

        assert_eq!(batches.len(), 1);
        assert_eq!(ResultSet::row_count_of(&batches), 2);
    }

    #[tokio::test]
    async fn test_fetch_all_without_handle_returns_buffered_batches_only() {
        let mut transport = MockTransport::new();
        transport.expect_fetch_results().times(0);
        transport.expect_close_result_set().times(0);

        let batches = streaming_result_set(transport, &[1, 2, 3], 3, None)
            .fetch_all()
            .await
            .unwrap();

        assert_eq!(batches.len(), 1);
        assert_eq!(ResultSet::row_count_of(&batches), 3);
    }

    #[tokio::test]
    async fn test_fetch_all_without_handle_fails_when_buffered_rows_miss_total() {
        let mut transport = MockTransport::new();
        transport.expect_fetch_results().times(0);
        transport.expect_close_result_set().times(0);

        let err = streaming_result_set(transport, &[1], 5, None)
            .fetch_all()
            .await
            .unwrap_err();

        assert!(matches!(err, QueryError::ExecutionFailed(_)));
        let message = err.to_string();
        assert!(message.contains("delivered 1 rows"), "{message}");
        assert!(message.contains("total row count is 5"), "{message}");
    }

    #[tokio::test]
    async fn test_fetch_all_without_handle_fails_when_buffered_rows_exceed_total() {
        let mut transport = MockTransport::new();
        transport.expect_fetch_results().times(0);
        transport.expect_close_result_set().times(0);

        let err = streaming_result_set(transport, &[1, 2, 3], 2, None)
            .fetch_all()
            .await
            .unwrap_err();

        assert!(matches!(err, QueryError::ExecutionFailed(_)));
        let message = err.to_string();
        assert!(message.contains("delivered 3 rows"), "{message}");
        assert!(message.contains("total row count is 2"), "{message}");
    }

    #[tokio::test]
    async fn test_fetch_all_propagates_fetch_error() {
        let mut transport = MockTransport::new();
        transport.expect_fetch_results().returning(|_| {
            Err(TransportError::ProtocolError(
                "page fetch failed".to_string(),
            ))
        });
        transport
            .expect_close_result_set()
            .times(1)
            .returning(|_| Ok(()));

        let err = streaming_result_set(transport, &[1], 0, Some(ResultSetHandle::new(1)))
            .fetch_all()
            .await
            .unwrap_err();

        assert!(matches!(err, QueryError::ExecutionFailed(_)));
        assert!(err.to_string().contains("page fetch failed"));
    }

    #[tokio::test]
    async fn test_fetch_all_ignores_close_result_set_error() {
        let mut transport = MockTransport::new();
        transport
            .expect_close_result_set()
            .times(1)
            .returning(|_| Err(TransportError::IoError("socket gone".to_string())));

        let batches = streaming_result_set(transport, &[1], 1, Some(ResultSetHandle::new(1)))
            .fetch_all()
            .await
            .unwrap();

        assert_eq!(batches.len(), 1);
    }

    /// Scenario: Small result set retrieval
    #[tokio::test]
    async fn test_fetch_all_small_result_set_sends_no_fetch() {
        let mut transport = MockTransport::new();
        transport.expect_fetch_results().times(0);
        transport.expect_close_result_set().times(0);

        let batches = streaming_result_set(transport, &[1, 2], 2, None)
            .fetch_all()
            .await
            .unwrap();

        assert_eq!(ResultSet::row_count_of(&batches), 2);
    }

    /// Scenario: Result set that ends before its total row count fails
    #[tokio::test]
    async fn test_fetch_all_fails_when_the_stream_ends_before_the_total() {
        let mut transport = MockTransport::new();
        transport
            .expect_fetch_results()
            .times(1)
            .returning(|_| Ok(single_column_result_data(&[], 3)));
        transport
            .expect_close_result_set()
            .times(1)
            .returning(|_| Ok(()));

        let err = streaming_result_set(transport, &[1], 3, Some(ResultSetHandle::new(1)))
            .fetch_all()
            .await
            .unwrap_err();

        assert!(matches!(err, QueryError::ExecutionFailed(_)));
        let message = err.to_string();
        assert!(message.contains('1') && message.contains('3'), "{message}");
    }

    /// Scenario: Result set that exceeds its total row count fails
    #[tokio::test]
    async fn test_fetch_all_fails_when_a_page_exceeds_the_total() {
        let mut transport = MockTransport::new();
        transport
            .expect_fetch_results()
            .times(1)
            .returning(|_| Ok(single_column_result_data(&[2, 3], 2)));
        transport
            .expect_close_result_set()
            .times(1)
            .returning(|_| Ok(()));

        let err = streaming_result_set(transport, &[1], 2, Some(ResultSetHandle::new(1)))
            .fetch_all()
            .await
            .unwrap_err();

        assert!(matches!(err, QueryError::ExecutionFailed(_)));
        let message = err.to_string();
        assert!(message.contains('3') && message.contains('2'), "{message}");
    }

    #[tokio::test]
    async fn test_fetch_all_on_row_count_result_reports_no_result_set() {
        let transport: Arc<Mutex<dyn TransportProtocol>> =
            Arc::new(Mutex::new(MockTransport::new()));
        let result_set = ResultSet::from_transport_result(
            TransportQueryResult::RowCount { count: 5 },
            transport,
        )
        .unwrap();

        let err = result_set.fetch_all().await.unwrap_err();

        assert!(matches!(err, QueryError::NoResultSet(_)));
        assert!(err.to_string().contains("Cannot fetch batches"));
    }

    #[test]
    fn test_into_iterator_on_row_count_result_reports_no_result_set() {
        let transport: Arc<Mutex<dyn TransportProtocol>> =
            Arc::new(Mutex::new(MockTransport::new()));
        let result_set = ResultSet::from_transport_result(
            TransportQueryResult::RowCount { count: 5 },
            transport,
        )
        .unwrap();

        let err = match result_set.into_iterator() {
            Err(err) => err,
            Ok(_) => panic!("row count result must not yield an iterator"),
        };

        assert!(matches!(err, QueryError::NoResultSet(_)));
        assert!(err.to_string().contains("Cannot iterate"));
    }

    // =========================================================================
    // Tests for Debug on ResultSet
    // =========================================================================

    #[test]
    fn test_result_set_debug_redacts_transport() {
        let rendered = format!(
            "{:?}",
            streaming_result_set(MockTransport::new(), &[1], 1, None)
        );

        assert!(rendered.starts_with("ResultSet {"), "got: {}", rendered);
        assert!(rendered.contains("Stream"));
        assert!(rendered.contains("<TransportProtocol>"));
    }

    #[test]
    fn test_result_set_debug_shows_row_count_variant() {
        let transport: Arc<Mutex<dyn TransportProtocol>> =
            Arc::new(Mutex::new(MockTransport::new()));
        let result_set = ResultSet::from_transport_result(
            TransportQueryResult::RowCount { count: 42 },
            transport,
        )
        .unwrap();

        let rendered = format!("{:?}", result_set);

        assert!(rendered.contains("RowCount"));
        assert!(rendered.contains("42"));
        assert!(rendered.contains("<TransportProtocol>"));
    }

    // =========================================================================
    // Tests for ResultSetIterator batch fetching
    // =========================================================================

    #[test]
    fn test_next_batch_serves_buffered_page_then_fetches_the_next() {
        let runtime = entered_runtime();
        let _guard = runtime.enter();

        let transport = transport_serving_one_more_page(&[2]);

        let mut iterator = streaming_iterator(transport, &[1], 0, Some(ResultSetHandle::new(1)));

        assert_eq!(iterator.next_batch().unwrap().unwrap().num_rows(), 1);
        assert_eq!(iterator.next_batch().unwrap().unwrap().num_rows(), 1);
        assert!(iterator.next_batch().is_none());
        // Once exhausted the iterator must not fetch again.
        assert!(iterator.next_batch().is_none());
    }

    #[test]
    fn test_next_batch_propagates_fetch_error() {
        let runtime = entered_runtime();
        let _guard = runtime.enter();

        let mut transport = MockTransport::new();
        transport
            .expect_fetch_results()
            .times(1)
            .returning(|_| Err(TransportError::ReceiveError("no page".to_string())));

        let mut iterator = streaming_iterator(transport, &[1], 0, Some(ResultSetHandle::new(1)));
        assert_eq!(iterator.next_batch().unwrap().unwrap().num_rows(), 1);

        let err = iterator.next_batch().unwrap().unwrap_err();

        assert!(matches!(err, QueryError::ExecutionFailed(_)));
        assert!(err.to_string().contains("no page"));
    }

    #[test]
    fn test_next_batch_fetches_again_after_a_transport_error() {
        let runtime = entered_runtime();
        let _guard = runtime.enter();

        let mut transport = MockTransport::new();
        let mut call = 0;
        transport
            .expect_fetch_results()
            .times(2)
            .returning(move |_| {
                call += 1;
                if call == 1 {
                    Err(TransportError::ReceiveError("no page".to_string()))
                } else {
                    Ok(single_column_result_data(&[2], 2))
                }
            });

        let mut iterator = streaming_iterator(transport, &[1], 2, Some(ResultSetHandle::new(1)));

        assert_eq!(iterator.next_batch().unwrap().unwrap().num_rows(), 1);
        assert!(matches!(
            iterator.next_batch().unwrap().unwrap_err(),
            QueryError::ExecutionFailed(_)
        ));
        assert_eq!(iterator.next_batch().unwrap().unwrap().num_rows(), 1);
        assert!(iterator.next_batch().is_none());
    }

    /// Scenario: Result set iterator ends after the last row
    #[test]
    fn test_next_batch_stops_at_total_rows_without_another_fetch() {
        let runtime = entered_runtime();
        let _guard = runtime.enter();

        let mut transport = MockTransport::new();
        transport
            .expect_fetch_results()
            .times(1)
            .returning(|_| Ok(single_column_result_data(&[2, 3], 3)));

        let mut iterator = streaming_iterator(transport, &[1], 3, Some(ResultSetHandle::new(1)));

        assert_eq!(iterator.next_batch().unwrap().unwrap().num_rows(), 1);
        assert_eq!(iterator.next_batch().unwrap().unwrap().num_rows(), 2);
        assert!(iterator.next_batch().is_none());
        assert!(iterator.next_batch().is_none());
    }

    /// Scenario: Result set that ends before its total row count fails
    #[test]
    fn test_next_batch_fails_when_the_stream_ends_before_the_total() {
        let runtime = entered_runtime();
        let _guard = runtime.enter();

        let mut transport = MockTransport::new();
        transport
            .expect_fetch_results()
            .times(1)
            .returning(|_| Ok(single_column_result_data(&[], 3)));

        let mut iterator = streaming_iterator(transport, &[1], 3, Some(ResultSetHandle::new(1)));
        assert_eq!(iterator.next_batch().unwrap().unwrap().num_rows(), 1);

        let err = iterator.next_batch().unwrap().unwrap_err();

        assert!(matches!(err, QueryError::ExecutionFailed(_)));
        let message = err.to_string();
        assert!(message.contains('1') && message.contains('3'), "{message}");
        assert!(iterator.next_batch().is_none());
    }

    /// Scenario: Result set that exceeds its total row count fails
    #[test]
    fn test_next_batch_fails_when_a_page_exceeds_the_total() {
        let runtime = entered_runtime();
        let _guard = runtime.enter();

        let mut transport = MockTransport::new();
        transport
            .expect_fetch_results()
            .times(1)
            .returning(|_| Ok(single_column_result_data(&[2, 3], 2)));

        let mut iterator = streaming_iterator(transport, &[1], 2, Some(ResultSetHandle::new(1)));
        assert_eq!(iterator.next_batch().unwrap().unwrap().num_rows(), 1);

        let err = iterator.next_batch().unwrap().unwrap_err();

        assert!(matches!(err, QueryError::ExecutionFailed(_)));
        let message = err.to_string();
        assert!(message.contains('3') && message.contains('2'), "{message}");
        assert!(iterator.next_batch().is_none());
    }

    #[tokio::test]
    async fn test_fetch_next_batch_short_circuits_once_the_stream_is_complete() {
        let mut transport = MockTransport::new();
        transport.expect_fetch_results().times(0);
        let transport: Arc<Mutex<dyn TransportProtocol>> = Arc::new(Mutex::new(transport));

        let mut iterator = ResultSetIterator {
            handle: Some(ResultSetHandle::new(1)),
            transport,
            metadata: QueryMetadata::new(
                ResultSet::build_schema(&[decimal_column("id")]).unwrap(),
                Some(1),
            ),
            batches: Vec::new(),
            current_index: 0,
            complete: true,
        };

        assert!(iterator.fetch_next_batch().await.unwrap().is_none());
    }

    #[test]
    fn test_next_batch_reports_exhausted_when_handle_was_already_released() {
        let runtime = entered_runtime();
        let _guard = runtime.enter();

        let mut transport = MockTransport::new();
        transport.expect_fetch_results().times(0);
        let transport: Arc<Mutex<dyn TransportProtocol>> = Arc::new(Mutex::new(transport));

        let mut iterator = ResultSetIterator {
            handle: None,
            transport,
            metadata: QueryMetadata::new(
                ResultSet::build_schema(&[decimal_column("id")]).unwrap(),
                Some(1),
            ),
            batches: Vec::new(),
            current_index: 0,
            complete: false,
        };

        assert!(iterator.next_batch().is_none());
    }

    #[test]
    fn test_next_batch_without_async_runtime_reports_invalid_state() {
        let mut transport = MockTransport::new();
        transport.expect_fetch_results().times(0);

        let mut iterator = streaming_iterator(transport, &[1], 0, Some(ResultSetHandle::new(1)));
        assert_eq!(iterator.next_batch().unwrap().unwrap().num_rows(), 1);

        let err = iterator.next_batch().unwrap().unwrap_err();

        assert!(matches!(err, QueryError::InvalidState(_)));
        assert!(err.to_string().contains("No async runtime available"));
    }

    #[test]
    fn test_iterator_trait_yields_every_page_until_exhausted() {
        let runtime = entered_runtime();
        let _guard = runtime.enter();

        let mut transport = MockTransport::new();
        transport
            .expect_fetch_results()
            .times(1)
            .returning(|_| Ok(single_column_result_data(&[], 0)));

        let iterator = streaming_iterator(transport, &[1, 2], 0, Some(ResultSetHandle::new(1)));
        let batches: Vec<RecordBatch> = iterator.collect::<Result<Vec<_>, _>>().unwrap();

        assert_eq!(batches.len(), 1);
        assert_eq!(ResultSet::row_count_of(&batches), 2);
    }

    // =========================================================================
    // Tests for ResultSetIterator::close
    // =========================================================================

    #[tokio::test]
    async fn test_iterator_close_releases_the_handle() {
        let handle = ResultSetHandle::new(9);
        let mut transport = MockTransport::new();
        transport
            .expect_close_result_set()
            .with(eq(handle))
            .times(1)
            .returning(|_| Ok(()));

        streaming_iterator(transport, &[1], 1, Some(handle))
            .close()
            .await
            .unwrap();
    }

    #[tokio::test]
    async fn test_iterator_close_without_handle_is_a_no_op() {
        let mut transport = MockTransport::new();
        transport.expect_close_result_set().times(0);

        streaming_iterator(transport, &[1], 1, None)
            .close()
            .await
            .unwrap();
    }

    #[tokio::test]
    async fn test_iterator_close_propagates_transport_error() {
        let mut transport = MockTransport::new();
        transport
            .expect_close_result_set()
            .times(1)
            .returning(|_| Err(TransportError::IoError("closed twice".to_string())));

        let err = streaming_iterator(transport, &[1], 1, Some(ResultSetHandle::new(1)))
            .close()
            .await
            .unwrap_err();

        assert!(matches!(err, QueryError::ExecutionFailed(_)));
        assert!(err.to_string().contains("closed twice"));
    }

    // =========================================================================
    // Tests for Arrow payload pass-through
    // =========================================================================

    fn decimal_batch(values: Vec<i128>) -> RecordBatch {
        let data_type = arrow::datatypes::DataType::Decimal128(18, 0);
        let schema = Arc::new(Schema::new(vec![Field::new("id", data_type.clone(), true)]));
        let column = arrow::array::Decimal128Array::from(values).with_data_type(data_type);
        RecordBatch::try_new(schema, vec![Arc::new(column)]).unwrap()
    }

    fn arrow_payload(batch: RecordBatch) -> ResultData {
        ResultData {
            columns: vec![],
            total_rows: batch.num_rows() as i64,
            data: ResultPayload::Arrow(batch),
        }
    }

    fn metadata_schema_of(columns: &[ColumnInfo]) -> Arc<Schema> {
        ResultSet::build_schema(columns).unwrap()
    }

    #[test]
    fn test_build_schema_marks_fields_nullable_and_carries_exasol_metadata() {
        let columns = vec![
            ColumnInfo {
                name: "AMOUNT".to_string(),
                data_type: DataType {
                    precision: Some(10),
                    scale: Some(2),
                    ..plain_type("DECIMAL")
                },
            },
            ColumnInfo {
                name: "NAME".to_string(),
                data_type: DataType {
                    size: Some(50),
                    ..plain_type("VARCHAR")
                },
            },
            ColumnInfo {
                name: "CODE".to_string(),
                data_type: DataType {
                    size: Some(3),
                    ..plain_type("CHAR")
                },
            },
        ];

        let schema = ResultSet::build_schema(&columns).unwrap();

        assert!(schema.fields().iter().all(|field| field.is_nullable()));
        let amount = schema.field(0);
        assert_eq!(
            amount.data_type(),
            &arrow::datatypes::DataType::Decimal128(10, 2)
        );
        assert_eq!(
            amount.metadata().get("exasol:type_name").unwrap(),
            "DECIMAL"
        );
        assert_eq!(amount.metadata().get("exasol:precision").unwrap(), "10");
        assert_eq!(amount.metadata().get("exasol:scale").unwrap(), "2");
        let name = schema.field(1);
        assert_eq!(name.metadata().get("exasol:type_name").unwrap(), "VARCHAR");
        assert_eq!(name.metadata().get("exasol:size").unwrap(), "50");
        let code = schema.field(2);
        assert_eq!(code.metadata().get("exasol:type_name").unwrap(), "CHAR");
        assert_eq!(code.metadata().get("exasol:size").unwrap(), "3");
        for field in schema.fields() {
            assert!(TypeMapper::from_field_metadata(field.metadata()).is_some());
        }
    }

    #[test]
    fn test_build_schema_fails_for_unsupported_column_type() {
        let columns = vec![ColumnInfo {
            name: "X".to_string(),
            data_type: plain_type("UNKNOWN_TYPE"),
        }];

        assert!(ResultSet::build_schema(&columns).is_err());
    }

    #[test]
    fn test_payload_to_record_batch_keeps_arrow_columns_and_adds_field_metadata() {
        let batch = decimal_batch(vec![7, 8]);
        let schema = metadata_schema_of(&[decimal_column("id")]);

        let converted =
            ResultSet::payload_to_record_batch(&arrow_payload(batch.clone()), &schema).unwrap();

        assert_eq!(converted.num_rows(), 2);
        let field = converted.schema().field(0).clone();
        assert_eq!(field.name(), "id");
        assert_eq!(
            field.data_type(),
            &arrow::datatypes::DataType::Decimal128(18, 0)
        );
        assert!(field.is_nullable());
        assert_eq!(field.metadata().get("exasol:type_name").unwrap(), "DECIMAL");
        assert_eq!(field.metadata().get("exasol:precision").unwrap(), "18");
        assert_eq!(field.metadata().get("exasol:scale").unwrap(), "0");
        assert!(Arc::ptr_eq(converted.column(0), batch.column(0)));
    }

    #[test]
    fn test_payload_to_record_batch_keeps_timestamp_column_under_local_time_zone_schema_field() {
        use arrow::datatypes::{DataType as ArrowType, TimeUnit};

        let batch_type = ArrowType::Timestamp(TimeUnit::Microsecond, None);
        let batch_schema = Arc::new(Schema::new(vec![Field::new("T", batch_type.clone(), true)]));
        let batch = RecordBatch::try_new(
            batch_schema,
            vec![Arc::new(arrow::array::TimestampMicrosecondArray::from(
                vec![42],
            ))],
        )
        .unwrap();
        let schema = metadata_schema_of(&[ColumnInfo {
            name: "T".to_string(),
            data_type: DataType {
                with_local_time_zone: Some(true),
                ..plain_type("TIMESTAMP")
            },
        }]);
        assert_eq!(
            schema.field(0).data_type(),
            &ArrowType::Timestamp(TimeUnit::Microsecond, Some("UTC".into()))
        );

        let converted =
            ResultSet::payload_to_record_batch(&arrow_payload(batch.clone()), &schema).unwrap();

        let field = converted.schema().field(0).clone();
        assert_eq!(field.data_type(), &batch_type);
        assert_eq!(
            field.metadata().get("exasol:type_name").unwrap(),
            "TIMESTAMP WITH LOCAL TIME ZONE"
        );
        assert!(Arc::ptr_eq(converted.column(0), batch.column(0)));
    }

    fn json_batch_of(
        type_name: &str,
        values: Vec<serde_json::Value>,
    ) -> Result<RecordBatch, ConversionError> {
        let schema = metadata_schema_of(&[ColumnInfo {
            name: "IV_COL".to_string(),
            data_type: plain_type(type_name),
        }]);
        let rows: Vec<Vec<serde_json::Value>> = values.into_iter().map(|v| vec![v]).collect();
        ResultSet::column_major_to_record_batch(&rows, &schema)
    }

    #[test]
    fn test_json_interval_columns_convert_to_month_day_nano() {
        use arrow::array::IntervalMonthDayNanoArray;
        use arrow::datatypes::IntervalMonthDayNano as Imdn;

        let year_to_month =
            json_batch_of("INTERVAL YEAR TO MONTH", vec![json!("+01-02"), json!(null)]).unwrap();
        let array = year_to_month
            .column(0)
            .as_any()
            .downcast_ref::<IntervalMonthDayNanoArray>()
            .unwrap();
        assert_eq!(array.value(0), Imdn::new(14, 0, 0));
        assert!(array.is_null(1));

        let day_to_second =
            json_batch_of("INTERVAL DAY TO SECOND", vec![json!("-03 04:05:06.789")]).unwrap();
        let array = day_to_second
            .column(0)
            .as_any()
            .downcast_ref::<IntervalMonthDayNanoArray>()
            .unwrap();
        assert_eq!(array.value(0), Imdn::new(0, -3, -14_706_789_000_000));
    }

    #[test]
    fn test_json_interval_beyond_month_range_fails_naming_the_column() {
        let error = json_batch_of("INTERVAL YEAR TO MONTH", vec![json!("+999999999-11")])
            .unwrap_err()
            .to_string();

        assert!(error.contains("IV_COL"), "{error}");
        assert!(error.contains("+999999999-11"), "{error}");
    }

    #[test]
    fn test_json_interval_that_is_not_a_string_fails() {
        assert!(json_batch_of("INTERVAL DAY TO SECOND", vec![json!(5)]).is_err());
    }

    #[test]
    fn test_json_geometry_and_hashtype_columns_keep_text_bytes_as_binary() {
        use arrow::array::BinaryArray;

        for (type_name, text) in [
            ("GEOMETRY", "POINT (1 2)"),
            ("HASHTYPE", "550e8400e29b11d4a716446655440000"),
        ] {
            let batch = json_batch_of(type_name, vec![json!(text), json!(null)]).unwrap();
            let array = batch
                .column(0)
                .as_any()
                .downcast_ref::<BinaryArray>()
                .unwrap();
            assert_eq!(array.value(0), text.as_bytes());
            assert!(array.is_null(1));
        }
    }

    #[test]
    fn test_exasol_type_of_reads_the_geometry_srid() {
        let with_srid = crate::transport::messages::DataType {
            srid: Some(4326),
            ..plain_type("GEOMETRY")
        };
        assert_eq!(
            exasol_type_of(&with_srid).unwrap(),
            ExasolType::Geometry { srid: Some(4326) }
        );
        assert_eq!(
            exasol_type_of(&plain_type("GEOMETRY")).unwrap(),
            ExasolType::Geometry { srid: None }
        );
    }

    #[test]
    fn test_data_type_deserializes_the_geometry_srid() {
        let data_type: crate::transport::messages::DataType =
            serde_json::from_str(r#"{"type":"GEOMETRY","size":2000000,"srid":4326}"#).unwrap();

        assert_eq!(data_type.srid, Some(4326));
        assert_eq!(data_type.size, Some(2000000));
    }

    #[test]
    fn test_payload_to_record_batch_keeps_batch_field_nullability() {
        let batch_schema = Arc::new(Schema::new(vec![Field::new(
            "id",
            arrow::datatypes::DataType::Int64,
            false,
        )]));
        let batch = RecordBatch::try_new(
            batch_schema,
            vec![Arc::new(arrow::array::Int64Array::from(vec![1]))],
        )
        .unwrap();
        let schema = metadata_schema_of(&[decimal_column("id")]);

        let converted = ResultSet::payload_to_record_batch(&arrow_payload(batch), &schema).unwrap();

        assert!(!converted.schema().field(0).is_nullable());
    }

    #[test]
    fn test_payload_to_record_batch_keeps_row_count_of_batch_without_columns() {
        let options = RecordBatchOptions::new().with_row_count(Some(3));
        let batch =
            RecordBatch::try_new_with_options(Arc::new(Schema::empty()), vec![], &options).unwrap();
        let schema = metadata_schema_of(&[]);

        let converted = ResultSet::payload_to_record_batch(&arrow_payload(batch), &schema).unwrap();

        assert_eq!(converted.num_rows(), 3);
        assert_eq!(converted.num_columns(), 0);
    }

    #[test]
    fn test_payload_to_record_batch_fails_when_column_counts_differ() {
        let batch = decimal_batch(vec![1]);
        let schema = metadata_schema_of(&[decimal_column("a"), varchar_column("b")]);

        let error = ResultSet::payload_to_record_batch(&arrow_payload(batch), &schema).unwrap_err();

        assert!(
            matches!(error, ConversionError::SchemaMismatch(_)),
            "{error:?}"
        );
    }

    #[tokio::test]
    async fn test_fetched_arrow_page_carries_the_result_set_field_metadata() {
        let handle = ResultSetHandle::new(1);
        let mut transport = MockTransport::new();
        transport
            .expect_fetch_results()
            .times(1)
            .returning(|_| Ok(arrow_payload(decimal_batch(vec![2]))));
        transport
            .expect_close_result_set()
            .times(1)
            .returning(|_| Ok(()));

        let batches = streaming_result_set(transport, &[1], 2, Some(handle))
            .fetch_all()
            .await
            .unwrap();

        assert_eq!(batches.len(), 2);
        for batch in &batches {
            let field = batch.schema().field(0).clone();
            assert_eq!(field.metadata().get("exasol:type_name").unwrap(), "DECIMAL");
        }
    }

    #[tokio::test]
    async fn test_zero_row_batch_carries_the_field_metadata() {
        let transport: Arc<Mutex<dyn TransportProtocol>> =
            Arc::new(Mutex::new(MockTransport::new()));
        let data = ResultData {
            columns: vec![decimal_column("id")],
            data: ResultPayload::Json(vec![]),
            total_rows: 0,
        };

        let result_set = ResultSet::from_transport_result(
            TransportQueryResult::ResultSet { handle: None, data },
            transport,
        )
        .unwrap();
        let batches = result_set.fetch_all().await.unwrap();

        let field = batches[0].schema().field(0).clone();
        assert_eq!(field.metadata().get("exasol:type_name").unwrap(), "DECIMAL");
    }

    // =========================================================================
    // Remaining type-mapping and parsing branches
    // =========================================================================

    #[test]
    fn test_exasol_type_of_named_timestamp_with_local_time_zone() {
        // Exasol may spell the type out instead of setting the flag; both must
        // map to a timezone-carrying Arrow timestamp.
        let data_type = plain_type("TIMESTAMP WITH LOCAL TIME ZONE");

        let arrow_type = arrow_type_of(&data_type).unwrap();

        assert!(matches!(
            arrow_type,
            arrow::datatypes::DataType::Timestamp(_, Some(_))
        ));
    }

    #[test]
    fn test_column_major_to_record_batch_stamps_utc_on_zoned_timestamps() {
        let schema = Arc::new(Schema::new(vec![Field::new(
            "ts",
            arrow::datatypes::DataType::Timestamp(
                arrow::datatypes::TimeUnit::Microsecond,
                Some("UTC".into()),
            ),
            true,
        )]));
        let rows = vec![vec![json!("1970-01-01 00:00:01")], vec![Value::Null]];

        let batch = ResultSet::column_major_to_record_batch(&rows, &schema).unwrap();

        let column = batch
            .column(0)
            .as_any()
            .downcast_ref::<arrow::array::TimestampMicrosecondArray>()
            .expect("timestamp column");
        assert_eq!(column.value(0), 1_000_000);
        assert!(column.is_null(1));
        assert_eq!(
            batch.schema().field(0).data_type(),
            &arrow::datatypes::DataType::Timestamp(
                arrow::datatypes::TimeUnit::Microsecond,
                Some("UTC".into())
            )
        );
    }

    #[test]
    fn test_column_major_to_record_batch_decimal_from_unsupported_shape_is_null() {
        let schema = Arc::new(Schema::new(vec![Field::new(
            "amount",
            arrow::datatypes::DataType::Decimal128(18, 2),
            true,
        )]));
        let rows = vec![vec![Value::Null], vec![json!(true)], vec![json!([1, 2])]];

        let batch = ResultSet::column_major_to_record_batch(&rows, &schema).unwrap();

        assert_eq!(batch.num_rows(), 3);
        assert_eq!(batch.column(0).null_count(), 3);
    }

    #[test]
    fn test_parse_string_to_decimal_rejects_a_non_numeric_fractional_part() {
        let err = ResultSet::parse_string_to_decimal("1.ab", 2).unwrap_err();

        assert_eq!(err.reason, "Invalid decimal part: ab");
        assert!(!err.overflow);
    }

    #[test]
    fn test_parse_string_to_decimal_flags_an_i128_overflow() {
        let scaled_overflow =
            ResultSet::parse_string_to_decimal("99999999999999999999999999999999999999", 2)
                .unwrap_err();
        let integer_overflow = ResultSet::parse_string_to_decimal(&"9".repeat(40), 0).unwrap_err();

        assert!(scaled_overflow.overflow);
        assert!(integer_overflow.overflow);
        assert_eq!(
            integer_overflow.reason,
            format!("integer part {} exceeds the 128-bit range", "9".repeat(40))
        );
    }

    /// Scenario: Conversion error reporting
    #[test]
    fn json_decimal_conversion_error_names_the_column() {
        let schema = Arc::new(Schema::new(vec![Field::new(
            "AMOUNT",
            arrow::datatypes::DataType::Decimal128(10, 2),
            true,
        )]));
        let rows = vec![vec![json!("1.50")], vec![json!("1.2.3")]];

        let err = ResultSet::column_major_to_record_batch(&rows, &schema).unwrap_err();

        match err {
            ConversionError::ValueConversionFailed {
                row,
                column,
                column_name,
                value,
                target_type,
                message,
            } => {
                assert_eq!((row, column, column_name.as_str()), (1, 0, "AMOUNT"));
                assert_eq!(value, "\"1.2.3\"");
                assert_eq!(
                    target_type,
                    arrow::datatypes::DataType::Decimal128(10, 2).to_string()
                );
                assert_eq!(message, "Invalid decimal format: 1.2.3");
            }
            other => panic!("expected ValueConversionFailed, got {other:?}"),
        }
    }

    #[test]
    fn json_decimal_overflow_is_reported_as_numeric_overflow() {
        let schema = Arc::new(Schema::new(vec![Field::new(
            "AMOUNT",
            arrow::datatypes::DataType::Decimal128(38, 2),
            true,
        )]));
        let rows = vec![vec![json!("99999999999999999999999999999999999999")]];

        let err = ResultSet::column_major_to_record_batch(&rows, &schema).unwrap_err();

        assert!(matches!(
            err,
            ConversionError::NumericOverflow { row: 0, column: 0, ref column_name, .. }
                if column_name == "AMOUNT"
        ));
    }

    #[test]
    fn test_query_metadata_with_execution_time() {
        let schema = Arc::new(Schema::new(vec![Field::new(
            "id",
            arrow::datatypes::DataType::Int64,
            true,
        )]));

        let metadata = QueryMetadata::new(schema, Some(3)).with_execution_time(17);

        assert_eq!(metadata.execution_time_ms, Some(17));
        assert_eq!(metadata.total_rows, Some(3));
        assert_eq!(metadata.column_count, 1);
    }
}