google-cloud-bigquery 0.19.0

// Copyright 2026 Google LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     https://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#[cfg(any(
    google_cloud_unstable_bigquery_arrow,
    google_cloud_unstable_bigquery_storage_read
))]
pub(crate) use super::arrow::ArrowCell;
use base64::Engine;
use base64::prelude::BASE64_STANDARD;
#[allow(unused_imports)]
use wkt::{Struct, Timestamp, Value};

use crate::error::ConvertError;

pub(crate) const BIGQUERY_DATE_FORMAT: &[time::format_description::FormatItem<'static>] =
    time::macros::format_description!("[year]-[month]-[day]");
pub(crate) const BIGQUERY_TIME_FORMAT: &[time::format_description::FormatItem<'static>] = time::macros::format_description!(
    "[hour padding:none]:[minute padding:none]:[second padding:none]"
);
pub(crate) const BIGQUERY_TIME_SUBSEC_FORMAT: &[time::format_description::FormatItem<'static>] = time::macros::format_description!(
    "[hour padding:none]:[minute padding:none]:[second padding:none].[subsecond]"
);
pub(crate) const BIGQUERY_DATETIME_FORMAT: &[time::format_description::FormatItem<'static>] =
    time::macros::format_description!("[year]-[month]-[day]T[hour]:[minute]:[second]");
pub(crate) const BIGQUERY_DATETIME_SUBSEC_FORMAT: &[time::format_description::FormatItem<
    'static,
>] = time::macros::format_description!("[year]-[month]-[day]T[hour]:[minute]:[second].[subsecond]");

#[derive(Clone, Debug, PartialEq)]
pub(crate) enum SqlValueInner {
    Null,
    Bool(bool),
    Number(serde_json::Number),
    String(String),
    Array(Vec<SqlValueInner>),
    Struct(Vec<(String, SqlValueInner)>),
    #[cfg(any(
        google_cloud_unstable_bigquery_arrow,
        google_cloud_unstable_bigquery_storage_read
    ))]
    #[cfg_attr(not(test), allow(dead_code))]
    Arrow(ArrowCell),
}

impl SqlValueInner {
    pub(crate) fn type_name(&self) -> String {
        match self {
            Self::Null => "null".to_string(),
            Self::Bool(_) => "bool".to_string(),
            Self::Number(_) => "number".to_string(),
            Self::String(_) => "string".to_string(),
            Self::Array(_) => "array".to_string(),
            Self::Struct(_) => "object".to_string(),
            #[cfg(any(
                google_cloud_unstable_bigquery_arrow,
                google_cloud_unstable_bigquery_storage_read
            ))]
            Self::Arrow(cell) => cell.data_type_str(),
        }
    }

    pub(crate) fn from_wkt(value: wkt::Value) -> Self {
        match value {
            wkt::Value::Null => Self::Null,
            wkt::Value::Bool(b) => Self::Bool(b),
            wkt::Value::Number(n) => Self::Number(n),
            wkt::Value::String(s) => Self::String(s),
            wkt::Value::Array(arr) => Self::Array(arr.into_iter().map(Self::from_wkt).collect()),
            wkt::Value::Object(obj) => Self::Struct(
                obj.into_iter()
                    .map(|(k, v)| (k, Self::from_wkt(v)))
                    .collect(),
            ),
        }
    }
}

/// An opaque representation of a BigQuery SQL cell value.
///
/// Passed to [`FromSql::from_value`] when converting query result cells
/// into Rust types.
#[derive(Clone, Debug)]
pub struct SqlValue {
    pub(crate) inner: SqlValueInner,
    // Opts out of `UnwindSafe` and `RefUnwindSafe` while preserving `Send + Sync + Unpin`.
    _not_unwind_safe: std::marker::PhantomData<dyn Send + Sync + Unpin>,
}

impl SqlValue {
    pub(crate) fn from_inner(inner: SqlValueInner) -> Self {
        Self {
            inner,
            _not_unwind_safe: std::marker::PhantomData,
        }
    }

    /// Takes ownership of a field by name (`&str`) or zero-based index (`usize`)
    /// from a `STRUCT`, `ARRAY`, or `JSON` value and converts it to `T`.
    ///
    /// Used by the [`#[derive(FromSql)]`](derive@crate::query::FromSql) macro to
    /// extract struct fields.
    pub fn take<T: FromSql, I: crate::query::ColumnIndex>(
        &mut self,
        index: I,
    ) -> Result<T, ConvertError> {
        let val = index.take_sql_value(self)?;
        T::from_value(val)
    }
}

/// A trait for converting BigQuery value representations into Rust
/// types.
///
/// The BigQuery client uses [`wkt::Value`] by default, but will support
/// Arrow record batches via the Storage Read API integration in the future.
///
/// <div class="warning">
///
/// **Do not implement this trait directly.**
///
/// This trait is not intended for manual implementation. New methods (such as
/// Arrow conversion methods) may be added in future versions, which would break
/// manual implementations. To deserialize custom structs, use [`#[derive(FromSql)]`](derive@crate::query::FromSql)
/// or [`#[derive(FromRow)]`](crate::query::FromRow) instead.
///
/// </div>
///
/// [`Row::get()`](crate::query::Row::get) or [`Row::take()`](crate::query::Row::take)
/// use this trait to convert cell values, and the [`FromRow`](crate::query::FromRow)
/// derive macro uses it for field deserialization.
///
/// # Supported Types
///
/// Built-in implementations include:
/// - Numbers: `i32`, `i64`, `f32`, `f64`, [`Decimal`](rust_decimal::Decimal)
/// - Text & Bytes: `String`, `Vec<u8>` (decoded from base64), [`Bytes`](bytes::Bytes)
/// - Dates & Times: [`Timestamp`](wkt::Timestamp), [`Date`](google_cloud_type::model::Date), [`TimeOfDay`](google_cloud_type::model::TimeOfDay), [`DateTime`](google_cloud_type::model::DateTime)
/// - Intervals: [`Interval`](crate::datatypes::Interval)
/// - Collections: `Option<T>` (for `NULL`), `Vec<T>` (for repeated fields), [`Range<T>`](crate::datatypes::Range) (for `RANGE` types)
/// - Raw JSON: [`Value`](wkt::Value), [`Struct`](wkt::Struct)
///
/// # Example
///
/// ```
/// # use google_cloud_bigquery::client::BigQuery;
/// # async fn sample(client: BigQuery) -> anyhow::Result<()> {
/// let mut rows = client
///     .query("SELECT 12345 AS integer_col, 'foo' AS string_col")
///     .until_done()
///     .await?
///     .read();
///
/// while let Some(row) = rows.next().await.transpose()? {
///     let num: i64 = row.get("integer_col")?;
///     let txt: String = row.get("string_col")?;
///     println!("{txt}: {num}");
/// }
/// # Ok(())
/// # }
/// ```
pub trait FromSql: Sized {
    /// Converts a BigQuery [`SqlValue`] into the implementing type.
    fn from_value(value: SqlValue) -> Result<Self, ConvertError>;
}

impl FromSql for wkt::Value {
    fn from_value(value: SqlValue) -> Result<Self, ConvertError> {
        Ok(match value.inner {
            SqlValueInner::Null => wkt::Value::Null,
            SqlValueInner::Bool(b) => wkt::Value::Bool(b),
            SqlValueInner::Number(n) => wkt::Value::Number(n),
            SqlValueInner::String(s) => wkt::Value::String(s),
            SqlValueInner::Array(arr) => wkt::Value::Array(
                arr.into_iter()
                    .map(|v| wkt::Value::from_value(SqlValue::from_inner(v)))
                    .collect::<Result<Vec<_>, _>>()?,
            ),
            SqlValueInner::Struct(entries) => wkt::Value::Object(
                entries
                    .into_iter()
                    .map(|(k, v)| Ok((k, wkt::Value::from_value(SqlValue::from_inner(v))?)))
                    .collect::<Result<wkt::Struct, ConvertError>>()?,
            ),
            #[cfg(any(
                google_cloud_unstable_bigquery_arrow,
                google_cloud_unstable_bigquery_storage_read
            ))]
            SqlValueInner::Arrow(_) => {
                // TODO(#7032): Implement wkt::Value conversion for Arrow cells.
                return Err(ConvertError::Convert(
                    "Arrow to wkt::Value conversion is not yet implemented".into(),
                ));
            }
        })
    }
}

impl FromSql for String {
    fn from_value(value: SqlValue) -> Result<Self, ConvertError> {
        match value.inner {
            SqlValueInner::String(s) => Ok(s),
            SqlValueInner::Null => Err(ConvertError::NotNull),
            #[cfg(any(
                google_cloud_unstable_bigquery_arrow,
                google_cloud_unstable_bigquery_storage_read
            ))]
            SqlValueInner::Arrow(cell) => cell.as_str().map(str::to_owned),
            other => Err(ConvertError::type_mismatch("string", &other)),
        }
    }
}

impl FromSql for i32 {
    fn from_value(value: SqlValue) -> Result<Self, ConvertError> {
        match value.inner {
            SqlValueInner::Number(n) => n
                .as_i64()
                .and_then(|v| i32::try_from(v).ok())
                .ok_or_else(|| ConvertError::Convert("number is not a valid i32".into())),
            SqlValueInner::String(s) => s
                .parse::<i32>()
                .map_err(|e| ConvertError::Convert(Box::new(e))),
            SqlValueInner::Null => Err(ConvertError::NotNull),
            #[cfg(any(
                google_cloud_unstable_bigquery_arrow,
                google_cloud_unstable_bigquery_storage_read
            ))]
            SqlValueInner::Arrow(cell) => cell.as_i32(),
            other => Err(ConvertError::type_mismatch("number or string", &other)),
        }
    }
}

impl FromSql for i64 {
    fn from_value(value: SqlValue) -> Result<Self, ConvertError> {
        match value.inner {
            SqlValueInner::Number(n) => n
                .as_i64()
                .ok_or_else(|| ConvertError::Convert("number is not a valid i64".into())),
            SqlValueInner::String(s) => s
                .parse::<i64>()
                .map_err(|e| ConvertError::Convert(Box::new(e))),
            SqlValueInner::Null => Err(ConvertError::NotNull),
            #[cfg(any(
                google_cloud_unstable_bigquery_arrow,
                google_cloud_unstable_bigquery_storage_read
            ))]
            SqlValueInner::Arrow(cell) => cell.as_i64(),
            other => Err(ConvertError::type_mismatch("number or string", &other)),
        }
    }
}

impl FromSql for f32 {
    fn from_value(value: SqlValue) -> Result<Self, ConvertError> {
        match value.inner {
            SqlValueInner::Number(n) => n
                .as_f64()
                .map(|v| v as f32)
                .ok_or_else(|| ConvertError::Convert("number is not a valid f32".into())),
            SqlValueInner::String(s) => s
                .parse::<f32>()
                .map_err(|e| ConvertError::Convert(Box::new(e))),
            SqlValueInner::Null => Err(ConvertError::NotNull),
            #[cfg(any(
                google_cloud_unstable_bigquery_arrow,
                google_cloud_unstable_bigquery_storage_read
            ))]
            SqlValueInner::Arrow(cell) => cell.as_f32(),
            other => Err(ConvertError::type_mismatch("number or string", &other)),
        }
    }
}

impl FromSql for f64 {
    fn from_value(value: SqlValue) -> Result<Self, ConvertError> {
        match value.inner {
            SqlValueInner::Number(n) => n
                .as_f64()
                .ok_or_else(|| ConvertError::Convert("number is not a valid f64".into())),
            SqlValueInner::String(s) => s
                .parse::<f64>()
                .map_err(|e| ConvertError::Convert(Box::new(e))),
            SqlValueInner::Null => Err(ConvertError::NotNull),
            #[cfg(any(
                google_cloud_unstable_bigquery_arrow,
                google_cloud_unstable_bigquery_storage_read
            ))]
            SqlValueInner::Arrow(cell) => cell.as_f64(),
            other => Err(ConvertError::type_mismatch("number or string", &other)),
        }
    }
}

impl FromSql for bool {
    fn from_value(value: SqlValue) -> Result<Self, ConvertError> {
        match value.inner {
            SqlValueInner::Bool(b) => Ok(b),
            SqlValueInner::String(s) => s
                .parse::<bool>()
                .map_err(|e| ConvertError::Convert(Box::new(e))),
            SqlValueInner::Null => Err(ConvertError::NotNull),
            #[cfg(any(
                google_cloud_unstable_bigquery_arrow,
                google_cloud_unstable_bigquery_storage_read
            ))]
            SqlValueInner::Arrow(cell) => cell.as_bool(),
            other => Err(ConvertError::type_mismatch("bool or string", &other)),
        }
    }
}

impl<T: FromSql> FromSql for Option<T> {
    fn from_value(value: SqlValue) -> Result<Self, ConvertError> {
        match value.inner {
            SqlValueInner::Null => Ok(None),
            #[cfg(any(
                google_cloud_unstable_bigquery_arrow,
                google_cloud_unstable_bigquery_storage_read
            ))]
            SqlValueInner::Arrow(cell) if cell.is_null() => Ok(None),
            other => T::from_value(SqlValue::from_inner(other)).map(Some),
        }
    }
}

impl<T: FromSql> FromSql for Vec<T> {
    fn from_value(value: SqlValue) -> Result<Self, ConvertError> {
        match value.inner {
            SqlValueInner::Array(arr) => arr
                .into_iter()
                .map(|v| T::from_value(SqlValue::from_inner(v)))
                .collect(),
            SqlValueInner::Null => Err(ConvertError::NotNull),
            #[cfg(any(
                google_cloud_unstable_bigquery_arrow,
                google_cloud_unstable_bigquery_storage_read
            ))]
            SqlValueInner::Arrow(cell) => {
                let value_arr = cell.list_array_ref()?;
                (0..value_arr.len())
                    .map(|i| {
                        let elem_cell = ArrowCell::new(value_arr.clone(), i);
                        T::from_value(SqlValue::from_inner(SqlValueInner::Arrow(elem_cell)))
                    })
                    .collect()
            }
            other => Err(ConvertError::type_mismatch("array", &other)),
        }
    }
}

impl FromSql for wkt::Struct {
    fn from_value(value: SqlValue) -> Result<Self, ConvertError> {
        match value.inner {
            SqlValueInner::Struct(entries) => entries
                .into_iter()
                .map(|(k, v)| Ok((k, wkt::Value::from_value(SqlValue::from_inner(v))?)))
                .collect(),
            SqlValueInner::String(s) => {
                serde_json::from_str(&s).map_err(|e| ConvertError::Convert(Box::new(e)))
            }
            SqlValueInner::Null => Err(ConvertError::NotNull),
            #[cfg(any(
                google_cloud_unstable_bigquery_arrow,
                google_cloud_unstable_bigquery_storage_read
            ))]
            SqlValueInner::Arrow(cell) => {
                let s = cell.as_str()?;
                serde_json::from_str(s).map_err(|e| ConvertError::Convert(Box::new(e)))
            }
            other => Err(ConvertError::type_mismatch("object or string", &other)),
        }
    }
}

impl FromSql for wkt::Timestamp {
    fn from_value(value: SqlValue) -> Result<Self, ConvertError> {
        match value.inner {
            SqlValueInner::String(s) => {
                let micros = s
                    .parse::<i64>()
                    .map_err(|e| ConvertError::Convert(Box::new(e)))?;
                timestamp_from_micros(micros)
            }
            SqlValueInner::Number(n) => {
                let micros = n.as_i64().ok_or_else(|| {
                    ConvertError::Convert("timestamp number is not valid i64".into())
                })?;
                timestamp_from_micros(micros)
            }
            SqlValueInner::Null => Err(ConvertError::NotNull),
            #[cfg(any(
                google_cloud_unstable_bigquery_arrow,
                google_cloud_unstable_bigquery_storage_read
            ))]
            SqlValueInner::Arrow(cell) => {
                let micros = cell.as_timestamp_micros()?;
                timestamp_from_micros(micros)
            }
            other => Err(ConvertError::type_mismatch("string or number", &other)),
        }
    }
}

fn timestamp_from_micros(micros: i64) -> Result<wkt::Timestamp, ConvertError> {
    wkt::Timestamp::new(
        micros.div_euclid(1_000_000),
        (micros.rem_euclid(1_000_000) * 1_000) as i32,
    )
    .map_err(|e| ConvertError::Convert(Box::new(e)))
}

impl FromSql for google_cloud_type::model::Date {
    fn from_value(value: SqlValue) -> Result<Self, ConvertError> {
        match value.inner {
            SqlValueInner::String(s) => {
                let date = time::Date::parse(s.as_str(), BIGQUERY_DATE_FORMAT)
                    .map_err(|e| ConvertError::Convert(Box::new(e)))?;
                Ok(google_cloud_type::model::Date::new()
                    .set_year(date.year())
                    .set_month(u8::from(date.month()) as i32)
                    .set_day(date.day() as i32))
            }
            SqlValueInner::Null => Err(ConvertError::NotNull),
            #[cfg(any(
                google_cloud_unstable_bigquery_arrow,
                google_cloud_unstable_bigquery_storage_read
            ))]
            SqlValueInner::Arrow(cell) => {
                let days = cell.as_date32()?;
                let date = time::OffsetDateTime::from_unix_timestamp(i64::from(days) * 86_400)
                    .map_err(|e| ConvertError::Convert(Box::new(e)))?
                    .date();
                Ok(google_cloud_type::model::Date::new()
                    .set_year(date.year())
                    .set_month(i32::from(u8::from(date.month())))
                    .set_day(i32::from(date.day())))
            }
            other => Err(ConvertError::type_mismatch("string", &other)),
        }
    }
}

pub(crate) fn parse_time(s: &str) -> Result<time::Time, ConvertError> {
    let format = if s.contains('.') {
        BIGQUERY_TIME_SUBSEC_FORMAT
    } else {
        BIGQUERY_TIME_FORMAT
    };
    time::Time::parse(s, format).map_err(|e| ConvertError::Convert(Box::new(e)))
}

impl FromSql for google_cloud_type::model::TimeOfDay {
    fn from_value(value: SqlValue) -> Result<Self, ConvertError> {
        match value.inner {
            SqlValueInner::String(s) => {
                let time = parse_time(s.as_str())?;
                Ok(google_cloud_type::model::TimeOfDay::new()
                    .set_hours(time.hour() as i32)
                    .set_minutes(time.minute() as i32)
                    .set_seconds(time.second() as i32)
                    .set_nanos(time.nanosecond() as i32))
            }
            SqlValueInner::Null => Err(ConvertError::NotNull),
            #[cfg(any(
                google_cloud_unstable_bigquery_arrow,
                google_cloud_unstable_bigquery_storage_read
            ))]
            SqlValueInner::Arrow(cell) => {
                let micros = cell.as_time64_micros()?;
                if !(0..86_400_000_000).contains(&micros) {
                    return Err(ConvertError::Convert(
                        format!("time64 microseconds out of range: {micros}").into(),
                    ));
                }
                let nanos = (micros % 1_000_000) * 1_000;
                let total_secs = micros / 1_000_000;
                let seconds = total_secs % 60;
                let total_mins = total_secs / 60;
                let minutes = total_mins % 60;
                let hours = total_mins / 60;
                Ok(google_cloud_type::model::TimeOfDay::new()
                    .set_hours(hours as i32)
                    .set_minutes(minutes as i32)
                    .set_seconds(seconds as i32)
                    .set_nanos(nanos as i32))
            }
            other => Err(ConvertError::type_mismatch("string", &other)),
        }
    }
}

impl FromSql for google_cloud_type::model::DateTime {
    fn from_value(value: SqlValue) -> Result<Self, ConvertError> {
        match value.inner {
            SqlValueInner::String(s) => {
                let format = if s.contains('.') {
                    BIGQUERY_DATETIME_SUBSEC_FORMAT
                } else {
                    BIGQUERY_DATETIME_FORMAT
                };
                let dt = time::PrimitiveDateTime::parse(s.as_str(), format)
                    .map_err(|e| ConvertError::Convert(Box::new(e)))?;
                Ok(google_cloud_type::model::DateTime::new()
                    .set_year(dt.year())
                    .set_month(u8::from(dt.month()) as i32)
                    .set_day(dt.day() as i32)
                    .set_hours(dt.hour() as i32)
                    .set_minutes(dt.minute() as i32)
                    .set_seconds(dt.second() as i32)
                    .set_nanos(dt.nanosecond() as i32))
            }
            SqlValueInner::Null => Err(ConvertError::NotNull),
            #[cfg(any(
                google_cloud_unstable_bigquery_arrow,
                google_cloud_unstable_bigquery_storage_read
            ))]
            SqlValueInner::Arrow(cell) => {
                let micros = cell.as_timestamp_micros()?;
                let odt =
                    time::OffsetDateTime::from_unix_timestamp_nanos(i128::from(micros) * 1_000)
                        .map_err(|e| ConvertError::Convert(Box::new(e)))?;
                Ok(google_cloud_type::model::DateTime::new()
                    .set_year(odt.year())
                    .set_month(i32::from(u8::from(odt.month())))
                    .set_day(i32::from(odt.day()))
                    .set_hours(i32::from(odt.hour()))
                    .set_minutes(i32::from(odt.minute()))
                    .set_seconds(i32::from(odt.second()))
                    .set_nanos(odt.nanosecond() as i32))
            }
            other => Err(ConvertError::type_mismatch("string", &other)),
        }
    }
}

impl FromSql for google_cloud_type::model::Decimal {
    fn from_value(value: SqlValue) -> Result<Self, ConvertError> {
        match value.inner {
            SqlValueInner::String(s) => Ok(google_cloud_type::model::Decimal::new().set_value(s)),
            SqlValueInner::Number(n) => {
                Ok(google_cloud_type::model::Decimal::new().set_value(n.to_string()))
            }
            SqlValueInner::Null => Err(ConvertError::NotNull),
            #[cfg(any(
                google_cloud_unstable_bigquery_arrow,
                google_cloud_unstable_bigquery_storage_read
            ))]
            SqlValueInner::Arrow(cell) => cell
                .as_decimal_str()
                .map(|s| google_cloud_type::model::Decimal::new().set_value(s)),
            other => Err(ConvertError::type_mismatch("string or number", &other)),
        }
    }
}

impl FromSql for rust_decimal::Decimal {
    fn from_value(value: SqlValue) -> Result<Self, ConvertError> {
        match value.inner {
            SqlValueInner::String(s) => s
                .trim()
                .parse::<rust_decimal::Decimal>()
                .map_err(|e| ConvertError::Convert(Box::new(e))),
            SqlValueInner::Number(n) => {
                if let Some(i) = n.as_i64() {
                    Ok(rust_decimal::Decimal::from(i))
                } else if let Some(u) = n.as_u64() {
                    Ok(rust_decimal::Decimal::from(u))
                } else {
                    let f = n.as_f64().expect("Number must be i64, u64, or f64");
                    rust_decimal::Decimal::try_from(f)
                        .map_err(|e| ConvertError::Convert(Box::new(e)))
                }
            }
            SqlValueInner::Null => Err(ConvertError::NotNull),
            #[cfg(any(
                google_cloud_unstable_bigquery_arrow,
                google_cloud_unstable_bigquery_storage_read
            ))]
            SqlValueInner::Arrow(cell) => {
                if let Some((val, scale)) = cell.as_decimal128_with_scale()
                    && let Ok(d) = rust_decimal::Decimal::try_from_i128_with_scale(val, scale)
                {
                    return Ok(d.normalize());
                }
                cell.as_decimal_str()?
                    .parse::<rust_decimal::Decimal>()
                    .map(|d| d.normalize())
                    .map_err(|e| ConvertError::Convert(Box::new(e)))
            }
            other => Err(ConvertError::type_mismatch("string or number", &other)),
        }
    }
}

impl FromSql for Vec<u8> {
    fn from_value(value: SqlValue) -> Result<Self, ConvertError> {
        match value.inner {
            SqlValueInner::String(s) => BASE64_STANDARD
                .decode(s)
                .map_err(|e| ConvertError::Convert(Box::new(e))),
            SqlValueInner::Null => Err(ConvertError::NotNull),
            #[cfg(any(
                google_cloud_unstable_bigquery_arrow,
                google_cloud_unstable_bigquery_storage_read
            ))]
            SqlValueInner::Arrow(cell) => cell.as_bytes().map(<[u8]>::to_vec),
            other => Err(ConvertError::type_mismatch(
                "string (base64 encoded)",
                &other,
            )),
        }
    }
}

impl FromSql for bytes::Bytes {
    fn from_value(value: SqlValue) -> Result<Self, ConvertError> {
        Vec::<u8>::from_value(value).map(bytes::Bytes::from)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate as google_cloud_bigquery;
    use crate::query::FromSql;
    use google_cloud_type::model::Decimal;
    use rust_decimal::Decimal as RustDecimal;
    use test_case::test_case;

    impl SqlValue {
        pub(crate) fn new(value: wkt::Value) -> Self {
            Self::from_inner(SqlValueInner::from_wkt(value))
        }
    }

    // Test-only representation of `ConvertError` that implements `PartialEq`.
    // This allows testing error outcomes using `test_case` assertions without
    // implementing `PartialEq` on the production `ConvertError`.
    #[derive(Debug, PartialEq)]
    enum TestConvertError {
        NotNull,
        TypeMismatch(String),
        Convert,
        MissingField(String),
    }

    impl TestConvertError {
        fn type_mismatch(expected: &str) -> Self {
            Self::TypeMismatch(expected.to_string())
        }
    }

    impl From<ConvertError> for TestConvertError {
        fn from(err: ConvertError) -> Self {
            match err {
                ConvertError::NotNull => Self::NotNull,
                ConvertError::TypeMismatch { expected, .. } => Self::TypeMismatch(expected),
                ConvertError::Convert(_) => Self::Convert,
                ConvertError::MissingField(f) => Self::MissingField(f),
            }
        }
    }

    #[test]
    fn test_sql_value_traits() {
        static_assertions::assert_impl_all!(SqlValue: Send, Sync, Unpin, Clone, std::fmt::Debug);
        static_assertions::assert_not_impl_any!(
            SqlValue: std::panic::UnwindSafe,
            std::panic::RefUnwindSafe
        );
    }

    #[test_case(wkt::Value::Null => Ok(wkt::Value::Null) ; "value null")]
    #[test_case(wkt::Value::Bool(true) => Ok(wkt::Value::Bool(true)) ; "value bool")]
    #[test_case(wkt::Value::Number(42.into()) => Ok(wkt::Value::Number(42.into())) ; "value number")]
    #[test_case(wkt::Value::String("hello".to_string()) => Ok(wkt::Value::String("hello".to_string())) ; "value string")]
    #[test_case(wkt::Value::Array(vec![wkt::Value::Number(1.into()), wkt::Value::Bool(false)]) => Ok(wkt::Value::Array(vec![wkt::Value::Number(1.into()), wkt::Value::Bool(false)])) ; "value array")]
    #[test_case(wkt::Value::Object(wkt::Struct::from_iter([("k".to_string(), wkt::Value::String("v".to_string()))])) => Ok(wkt::Value::Object(wkt::Struct::from_iter([("k".to_string(), wkt::Value::String("v".to_string()))]))) ; "value struct")]
    fn test_from_sql_value(value: wkt::Value) -> Result<wkt::Value, TestConvertError> {
        FromSql::from_value(SqlValue::new(value)).map_err(TestConvertError::from)
    }

    #[derive(FromSql, Debug, PartialEq)]
    struct RawIdentSqlStruct {
        r#type: String,
        r#match: i64,
    }

    #[test]
    fn test_derive_from_sql_raw_identifier() {
        let val = wkt::Value::Object(wkt::Struct::from_iter([
            ("type".to_string(), wkt::Value::String("event".to_string())),
            ("match".to_string(), wkt::Value::Number(99.into())),
        ]));
        let parsed =
            RawIdentSqlStruct::from_value(SqlValue::new(val)).expect("should strip r# prefix");
        assert_eq!(
            parsed,
            RawIdentSqlStruct {
                r#type: "event".to_string(),
                r#match: 99,
            }
        );
    }

    #[test_case(wkt::Value::String("hello".to_string()) => Ok("hello".to_string()) ; "string")]
    #[test_case(wkt::Value::Null => Err(TestConvertError::NotNull) ; "null string")]
    #[test_case(wkt::Value::Number(123.into()) => Err(TestConvertError::type_mismatch("string")) ; "type mismatch string")]
    fn test_from_sql_string(value: wkt::Value) -> Result<String, TestConvertError> {
        FromSql::from_value(SqlValue::new(value)).map_err(TestConvertError::from)
    }

    #[test_case(wkt::Value::Number(123.into()) => Ok(123) ; "i64 from number")]
    #[test_case(wkt::Value::String("123".to_string()) => Ok(123) ; "i64 from string")]
    #[test_case(wkt::Value::Null => Err(TestConvertError::NotNull) ; "null i64")]
    #[test_case(wkt::Value::Bool(true) => Err(TestConvertError::type_mismatch("number or string")) ; "try bool as i64")]
    #[test_case(wkt::Value::String("hello".to_string()) => Err(TestConvertError::Convert) ; "invalid string as i64")]
    fn test_from_sql_i64(value: wkt::Value) -> Result<i64, TestConvertError> {
        FromSql::from_value(SqlValue::new(value)).map_err(TestConvertError::from)
    }

    #[test_case(wkt::Value::Number(serde_json::Number::from_f64(123.45).unwrap()) => Ok(123.45) ; "f64 from number")]
    #[test_case(wkt::Value::String("123.45".to_string()) => Ok(123.45) ; "f64 from string")]
    #[test_case(wkt::Value::Null => Err(TestConvertError::NotNull) ; "null f64")]
    #[test_case(wkt::Value::Bool(true) => Err(TestConvertError::type_mismatch("number or string")) ; "try bool as f64")]
    #[test_case(wkt::Value::String("hello".to_string()) => Err(TestConvertError::Convert) ; "invalid string as f64")]
    fn test_from_sql_f64(value: wkt::Value) -> Result<f64, TestConvertError> {
        FromSql::from_value(SqlValue::new(value)).map_err(TestConvertError::from)
    }

    #[test_case(wkt::Value::Bool(true) => Ok(true) ; "bool true")]
    #[test_case(wkt::Value::Bool(false) => Ok(false) ; "bool false")]
    #[test_case(wkt::Value::String("true".to_string()) => Ok(true) ; "bool from string true")]
    #[test_case(wkt::Value::String("false".to_string()) => Ok(false) ; "bool from string false")]
    #[test_case(wkt::Value::Null => Err(TestConvertError::NotNull) ; "null bool")]
    #[test_case(wkt::Value::Number(1.into()) => Err(TestConvertError::type_mismatch("bool or string")) ; "try number as bool")]
    #[test_case(wkt::Value::String("hello".to_string()) => Err(TestConvertError::Convert) ; "invalid string as bool")]
    fn test_from_sql_bool(value: wkt::Value) -> Result<bool, TestConvertError> {
        FromSql::from_value(SqlValue::new(value)).map_err(TestConvertError::from)
    }

    #[test_case(wkt::Value::Null => Ok(None) ; "option null")]
    #[test_case(wkt::Value::Number(123.into()) => Ok(Some(123)) ; "option some i64")]
    #[test_case(wkt::Value::String("hello".to_string()) => Err(TestConvertError::Convert) ; "option error i64")]
    fn test_from_sql_option(value: wkt::Value) -> Result<Option<i64>, TestConvertError> {
        FromSql::from_value(SqlValue::new(value)).map_err(TestConvertError::from)
    }

    #[test_case(wkt::Value::Array(vec![wkt::Value::Number(1.into()), wkt::Value::Number(2.into())]) => Ok(vec![1, 2]) ; "vec i64")]
    #[test_case(wkt::Value::Null => Err(TestConvertError::NotNull) ; "null vec i64")]
    #[test_case(wkt::Value::String("hello".to_string()) => Err(TestConvertError::type_mismatch("array")) ; "vec type mismatch")]
    #[test_case(wkt::Value::Array(vec![wkt::Value::String("invalid".to_string())]) => Err(TestConvertError::Convert) ; "vec element convert error")]
    fn test_from_sql_vec(value: wkt::Value) -> Result<Vec<i64>, TestConvertError> {
        FromSql::from_value(SqlValue::new(value)).map_err(TestConvertError::from)
    }

    #[test_case(wkt::Value::Object(wkt::Struct::from_iter([("a".to_string(), wkt::Value::Number(1.into()))])) => Ok(wkt::Struct::from_iter([("a".to_string(), wkt::Value::Number(1.into()))])) ; "struct ok")]
    #[test_case(wkt::Value::String(r#"{"a": 1}"#.to_string()) => Ok(wkt::Struct::from_iter([("a".to_string(), wkt::Value::Number(1.into()))])) ; "struct from json string")]
    #[test_case(wkt::Value::Null => Err(TestConvertError::NotNull) ; "struct null")]
    #[test_case(wkt::Value::String("hello".to_string()) => Err(TestConvertError::Convert) ; "struct invalid json string")]
    #[test_case(wkt::Value::Bool(true) => Err(TestConvertError::type_mismatch("object or string")) ; "struct type mismatch")]
    fn test_from_sql_struct(value: wkt::Value) -> Result<wkt::Struct, TestConvertError> {
        FromSql::from_value(SqlValue::new(value)).map_err(TestConvertError::from)
    }

    #[test_case(wkt::Value::String("1779982200000000".to_string()) => Ok(wkt::Timestamp::new(1779982200, 0).unwrap()) ; "timestamp micro integer string")]
    #[test_case(wkt::Value::Number(1779982200000000i64.into()) => Ok(wkt::Timestamp::new(1779982200, 0).unwrap()) ; "timestamp micro integer number")]
    #[test_case(wkt::Value::String("2026-05-28T15:30:00Z".to_string()) => Err(TestConvertError::Convert) ; "timestamp rfc3339 string fails")]
    #[test_case(wkt::Value::Number(serde_json::Number::from_f64(1779982200.5).unwrap()) => Err(TestConvertError::Convert) ; "timestamp f64 number fails")]
    #[test_case(wkt::Value::Null => Err(TestConvertError::NotNull) ; "timestamp null")]
    #[test_case(wkt::Value::Bool(true) => Err(TestConvertError::type_mismatch("string or number")) ; "timestamp type mismatch")]
    fn test_from_sql_timestamp(value: wkt::Value) -> Result<wkt::Timestamp, TestConvertError> {
        FromSql::from_value(SqlValue::new(value)).map_err(TestConvertError::from)
    }

    #[test_case(wkt::Value::String("2026-05-28".to_string()) => Ok(google_cloud_type::model::Date::new().set_year(2026).set_month(5).set_day(28)) ; "date valid")]
    #[test_case(wkt::Value::Null => Err(TestConvertError::NotNull) ; "null date")]
    #[test_case(wkt::Value::Number(123.into()) => Err(TestConvertError::type_mismatch("string")) ; "date type mismatch")]
    #[test_case(wkt::Value::String("invalid-date".to_string()) => Err(TestConvertError::Convert) ; "date invalid format")]
    #[test_case(wkt::Value::String("2026-abc-28".to_string()) => Err(TestConvertError::Convert) ; "date invalid digits")]
    fn test_from_sql_date(
        value: wkt::Value,
    ) -> Result<google_cloud_type::model::Date, TestConvertError> {
        FromSql::from_value(SqlValue::new(value)).map_err(TestConvertError::from)
    }

    #[test_case(wkt::Value::String("15:30:00".to_string()) => Ok(google_cloud_type::model::TimeOfDay::new().set_hours(15).set_minutes(30).set_seconds(0).set_nanos(0)) ; "time of day valid")]
    #[test_case(wkt::Value::String("15:30:00.123456".to_string()) => Ok(google_cloud_type::model::TimeOfDay::new().set_hours(15).set_minutes(30).set_seconds(0).set_nanos(123_456_000)) ; "time of day fractional")]
    #[test_case(wkt::Value::Null => Err(TestConvertError::NotNull) ; "time of day null")]
    #[test_case(wkt::Value::Number(123.into()) => Err(TestConvertError::type_mismatch("string")) ; "time of day type mismatch")]
    fn test_from_sql_time_of_day(
        value: wkt::Value,
    ) -> Result<google_cloud_type::model::TimeOfDay, TestConvertError> {
        FromSql::from_value(SqlValue::new(value)).map_err(TestConvertError::from)
    }

    #[test_case(wkt::Value::String("2026-05-28T15:30:00".to_string()) => Ok(google_cloud_type::model::DateTime::new().set_year(2026).set_month(5).set_day(28).set_hours(15).set_minutes(30).set_seconds(0).set_nanos(0)) ; "datetime without subseconds")]
    #[test_case(wkt::Value::String("2026-05-28T15:30:00.123456".to_string()) => Ok(google_cloud_type::model::DateTime::new().set_year(2026).set_month(5).set_day(28).set_hours(15).set_minutes(30).set_seconds(0).set_nanos(123_456_000)) ; "datetime with subseconds")]
    #[test_case(wkt::Value::Null => Err(TestConvertError::NotNull) ; "datetime null")]
    #[test_case(wkt::Value::Number(123.into()) => Err(TestConvertError::type_mismatch("string")) ; "datetime type mismatch")]
    fn test_from_sql_datetime(
        value: wkt::Value,
    ) -> Result<google_cloud_type::model::DateTime, TestConvertError> {
        FromSql::from_value(SqlValue::new(value)).map_err(TestConvertError::from)
    }

    #[test_case(wkt::Value::Number(123.into()) => Ok(123) ; "i32 from number")]
    #[test_case(wkt::Value::String("123".to_string()) => Ok(123) ; "i32 from string")]
    #[test_case(wkt::Value::Number(3_000_000_000i64.into()) => Err(TestConvertError::Convert) ; "i32 overflow from number")]
    #[test_case(wkt::Value::Null => Err(TestConvertError::NotNull) ; "null i32")]
    #[test_case(wkt::Value::Bool(true) => Err(TestConvertError::type_mismatch("number or string")) ; "try bool as i32")]
    #[test_case(wkt::Value::String("hello".to_string()) => Err(TestConvertError::Convert) ; "invalid string as i32")]
    fn test_from_sql_i32(value: wkt::Value) -> Result<i32, TestConvertError> {
        FromSql::from_value(SqlValue::new(value)).map_err(TestConvertError::from)
    }

    #[test_case(wkt::Value::Number(serde_json::Number::from_f64(123.45).unwrap()) => Ok(123.45) ; "f32 from number")]
    #[test_case(wkt::Value::String("123.45".to_string()) => Ok(123.45) ; "f32 from string")]
    #[test_case(wkt::Value::Null => Err(TestConvertError::NotNull) ; "null f32")]
    #[test_case(wkt::Value::Bool(true) => Err(TestConvertError::type_mismatch("number or string")) ; "try bool as f32")]
    #[test_case(wkt::Value::String("hello".to_string()) => Err(TestConvertError::Convert) ; "invalid string as f32")]
    fn test_from_sql_f32(value: wkt::Value) -> Result<f32, TestConvertError> {
        FromSql::from_value(SqlValue::new(value)).map_err(TestConvertError::from)
    }

    #[test_case(wkt::Value::String("123.456".to_string()) => Ok(Decimal::new().set_value("123.456")) ; "decimal from string")]
    #[test_case(wkt::Value::Number(serde_json::Number::from_f64(123.456).unwrap()) => Ok(Decimal::new().set_value("123.456")) ; "decimal from number")]
    #[test_case(wkt::Value::Null => Err(TestConvertError::NotNull) ; "null decimal")]
    #[test_case(wkt::Value::Bool(true) => Err(TestConvertError::type_mismatch("string or number")) ; "try bool as decimal")]
    fn test_from_sql_decimal(value: wkt::Value) -> Result<Decimal, TestConvertError> {
        FromSql::from_value(SqlValue::new(value)).map_err(TestConvertError::from)
    }

    #[test_case(wkt::Value::String("123.456".to_string()) => Ok(RustDecimal::from_str_exact("123.456").unwrap()) ; "rust_decimal from string")]
    #[test_case(wkt::Value::Number((-123i64).into()) => Ok(RustDecimal::from(-123i64)) ; "rust_decimal from i64 number")]
    #[test_case(wkt::Value::Number(u64::MAX.into()) => Ok(RustDecimal::from(u64::MAX)) ; "rust_decimal from u64 number")]
    #[test_case(wkt::Value::Number(serde_json::Number::from_f64(123.456).unwrap()) => Ok(RustDecimal::from_str_exact("123.456").unwrap()) ; "rust_decimal from f64 number")]
    #[test_case(wkt::Value::String("99999999999999999999999999999999.123".to_string()) => Err(TestConvertError::Convert) ; "rust_decimal overflow")]
    #[test_case(wkt::Value::Null => Err(TestConvertError::NotNull) ; "null rust_decimal")]
    #[test_case(wkt::Value::Bool(true) => Err(TestConvertError::type_mismatch("string or number")) ; "try bool as rust_decimal")]
    fn test_from_sql_rust_decimal(value: wkt::Value) -> Result<RustDecimal, TestConvertError> {
        FromSql::from_value(SqlValue::new(value)).map_err(TestConvertError::from)
    }

    #[test_case(wkt::Value::String("AQIDBA==".to_string()) => Ok(vec![1, 2, 3, 4]) ; "vec u8 from base64")]
    #[test_case(wkt::Value::String("".to_string()) => Ok(vec![]) ; "vec u8 from empty base64")]
    #[test_case(wkt::Value::Null => Err(TestConvertError::NotNull) ; "null vec u8")]
    #[test_case(wkt::Value::Bool(true) => Err(TestConvertError::type_mismatch("string (base64 encoded)")) ; "try bool as vec u8")]
    fn test_from_sql_vec_u8(value: wkt::Value) -> Result<Vec<u8>, TestConvertError> {
        FromSql::from_value(SqlValue::new(value)).map_err(TestConvertError::from)
    }

    #[test_case(wkt::Value::String("AQIDBA==".to_string()) => Ok(bytes::Bytes::from_static(&[1, 2, 3, 4])) ; "bytes from base64")]
    #[test_case(wkt::Value::String("".to_string()) => Ok(bytes::Bytes::from_static(&[])) ; "bytes from empty base64")]
    #[test_case(wkt::Value::Null => Err(TestConvertError::NotNull) ; "null bytes")]
    #[test_case(wkt::Value::Bool(true) => Err(TestConvertError::type_mismatch("string (base64 encoded)")) ; "try bool as bytes")]
    fn test_from_sql_bytes(value: wkt::Value) -> Result<bytes::Bytes, TestConvertError> {
        FromSql::from_value(SqlValue::new(value)).map_err(TestConvertError::from)
    }

    #[test_case("AQIDBA" ; "missing padding")]
    #[test_case("Not a base64 string" ; "words with spaces")]
    fn test_from_sql_bytes_invalid_base64(input: &str) {
        let err = bytes::Bytes::from_value(SqlValue::new(wkt::Value::String(input.to_string())))
            .unwrap_err();
        assert!(matches!(err, ConvertError::Convert(_)));

        let err = Vec::<u8>::from_value(SqlValue::new(wkt::Value::String(input.to_string())))
            .unwrap_err();
        assert!(matches!(err, ConvertError::Convert(_)));
    }

    #[derive(FromSql, Debug, PartialEq)]
    struct TestSqlStruct {
        name: String,
        #[bigquery(rename = "custom_int")]
        some_int: i64,
        some_bool: bool,
    }

    #[test_case(wkt::Value::Object(wkt::Struct::from_iter([("name".to_string(), wkt::Value::String("James".to_string())), ("custom_int".to_string(), wkt::Value::Number(272793.into())), ("some_bool".to_string(), wkt::Value::Bool(true))])) => Ok(TestSqlStruct { name: "James".to_string(), some_int: 272793, some_bool: true }) ; "object success")]
    #[test_case(wkt::Value::String(r#"{"name": "James", "custom_int": 272793, "some_bool": true}"#.to_string()) => Ok(TestSqlStruct { name: "James".to_string(), some_int: 272793, some_bool: true }) ; "json string success")]
    #[test_case(wkt::Value::Object(wkt::Struct::from_iter([("name".to_string(), wkt::Value::String("James".to_string())), ("some_bool".to_string(), wkt::Value::Bool(true))])) => Err(TestConvertError::MissingField("custom_int".to_string())) ; "missing field")]
    #[test_case(wkt::Value::Null => Err(TestConvertError::NotNull) ; "null error")]
    #[test_case(wkt::Value::String("invalid".to_string()) => Err(TestConvertError::Convert) ; "invalid json string")]
    #[test_case(wkt::Value::Bool(true) => Err(TestConvertError::type_mismatch("object or string")) ; "type mismatch")]
    fn test_derive_from_sql(value: wkt::Value) -> Result<TestSqlStruct, TestConvertError> {
        FromSql::from_value(SqlValue::new(value)).map_err(TestConvertError::from)
    }

    #[derive(FromSql, Debug, PartialEq)]
    struct ShadowedSqlStruct {
        iter: i64,
        obj: i64,
        arr: i64,
        value: String,
    }

    #[test]
    fn test_derive_from_sql_shadowing_field_names() {
        let from_obj = ShadowedSqlStruct::from_value(SqlValue::new(wkt::Value::Object(
            wkt::Struct::from_iter([
                ("iter".to_string(), wkt::Value::Number(10.into())),
                ("obj".to_string(), wkt::Value::Number(20.into())),
                ("arr".to_string(), wkt::Value::Number(30.into())),
                ("value".to_string(), wkt::Value::String("world".to_string())),
            ]),
        )))
        .expect("should deserialize from object");
        assert_eq!(
            from_obj,
            ShadowedSqlStruct {
                iter: 10,
                obj: 20,
                arr: 30,
                value: "world".to_string(),
            }
        );
    }

    #[test]
    fn test_sql_value_take_by_index_and_name() -> anyhow::Result<()> {
        let mut arr_val = SqlValue::new(wkt::Value::Array(vec![
            wkt::Value::Number(42.into()),
            wkt::Value::String("hello".to_string()),
        ]));
        assert_eq!(arr_val.take::<i64, _>(0)?, 42);
        assert_eq!(arr_val.take::<String, _>(1)?, "hello");
        assert!(matches!(
            arr_val.take::<i64, _>(2),
            Err(ConvertError::MissingField(ref f)) if f == "2"
        ));

        // Taking "foo" (at index 0) must replace slot 0 with Null in-place
        // without shifting index 1 ("bar") down to index 0.
        let mut struct_val = SqlValue::from_inner(SqlValueInner::Struct(vec![
            ("foo".to_string(), SqlValueInner::Number(10.into())),
            (
                "bar".to_string(),
                SqlValueInner::String("twenty".to_string()),
            ),
        ]));
        assert_eq!(struct_val.take::<i64, _>("foo".to_string())?, 10);
        // Subsequent reads of "foo" see Null (returning Ok(None) for Option<T>, matching Row::take)
        assert_eq!(struct_val.take::<Option<i64>, _>("foo")?, None);
        // Index 1 is still "bar" (not shifted)
        assert_eq!(struct_val.take::<String, _>(1)?, "twenty");

        // Positional take(usize) on a JSON array string
        let mut json_arr = SqlValue::new(wkt::Value::String(r#"[7, "eight"]"#.to_string()));
        assert_eq!(json_arr.take::<i64, _>(0)?, 7);
        assert_eq!(json_arr.take::<String, _>(1)?, "eight");

        // Invalid JSON string with positional take(usize)
        let mut invalid_json_arr = SqlValue::new(wkt::Value::String("not-json".to_string()));
        assert!(matches!(
            invalid_json_arr.take::<i64, _>(0),
            Err(ConvertError::Convert(_))
        ));

        // Positional take(usize) on Null and primitive Bool
        let mut null_val = SqlValue::new(wkt::Value::Null);
        assert!(matches!(
            null_val.take::<i64, _>(0),
            Err(ConvertError::NotNull)
        ));

        let mut bool_val = SqlValue::new(wkt::Value::Bool(true));
        assert!(matches!(
            bool_val.take::<i64, _>(0),
            Err(ConvertError::TypeMismatch { ref expected, ref got })
                if expected == "struct, array, or string" && got == "bool"
        ));

        // Cover SqlValueInner::type_name for Null, Array, and Struct
        assert_eq!(SqlValueInner::Null.type_name(), "null");
        assert_eq!(SqlValueInner::Array(vec![]).type_name(), "array");
        assert_eq!(SqlValueInner::Struct(vec![]).type_name(), "object");

        Ok(())
    }

    #[derive(FromSql, Debug, PartialEq)]
    struct GenericSql<T: Clone + Default, U: std::fmt::Debug> {
        #[bigquery(rename = "custom_val")]
        single: T,
        optional: Option<T>,
        list: Vec<T>,
        nested: NestedGenericSql<U>,
    }

    #[derive(FromSql, Debug, PartialEq)]
    struct NestedGenericSql<U> {
        inner_val: U,
    }

    #[test]
    fn test_derive_from_sql_generic() {
        let nested_obj = wkt::Value::Object(wkt::Struct::from_iter([(
            "inner_val".to_string(),
            wkt::Value::String("hello".to_string()),
        )]));
        let from_obj = GenericSql::<i64, String>::from_value(SqlValue::new(wkt::Value::Object(
            wkt::Struct::from_iter([
                ("custom_val".to_string(), wkt::Value::Number(100.into())),
                ("optional".to_string(), wkt::Value::Null),
                (
                    "list".to_string(),
                    wkt::Value::Array(vec![
                        wkt::Value::Number(1.into()),
                        wkt::Value::Number(2.into()),
                        wkt::Value::Number(3.into()),
                    ]),
                ),
                ("nested".to_string(), nested_obj),
            ]),
        )))
        .expect("should deserialize generic from object");

        assert_eq!(
            from_obj,
            GenericSql {
                single: 100,
                optional: None,
                list: vec![1, 2, 3],
                nested: NestedGenericSql {
                    inner_val: "hello".to_string(),
                },
            }
        );
    }

    #[derive(FromSql, Debug, PartialEq)]
    struct GenericSqlWhere<T>
    where
        T: std::fmt::Debug + Clone,
    {
        val: T,
    }

    #[test]
    fn test_derive_from_sql_generic_where_clause() {
        let val = GenericSqlWhere::<String>::from_value(SqlValue::new(wkt::Value::Object(
            wkt::Struct::from_iter([("val".to_string(), wkt::Value::String("hello".to_string()))]),
        )))
        .expect("should deserialize generic with where clause");

        assert_eq!(
            val,
            GenericSqlWhere {
                val: "hello".to_string(),
            }
        );
    }

    #[derive(FromSql, Debug, PartialEq)]
    struct GenericSqlDefault<T = i64> {
        val: T,
    }

    #[test]
    fn test_derive_from_sql_generic_default_param() {
        let val: GenericSqlDefault =
            GenericSqlDefault::from_value(SqlValue::new(wkt::Value::Object(
                wkt::Struct::from_iter([("val".to_string(), wkt::Value::Number(42.into()))]),
            )))
            .expect("should deserialize generic with default type param");

        assert_eq!(val, GenericSqlDefault { val: 42 });
    }

    #[derive(FromSql, Debug, PartialEq)]
    struct GenericTupleSql<T, U>(T, Option<T>, U);

    #[test]
    fn test_derive_from_sql_generic_tuple_struct() {
        let struct_val = SqlValue::from_inner(SqlValueInner::Struct(vec![
            ("".to_string(), SqlValueInner::Number(100.into())),
            ("".to_string(), SqlValueInner::Null),
            ("".to_string(), SqlValueInner::String("world".to_string())),
        ]));
        let parsed = GenericTupleSql::<i64, String>::from_value(struct_val)
            .expect("should deserialize generic tuple struct from struct");
        assert_eq!(parsed, GenericTupleSql(100, None, "world".to_string()));

        let json_val = SqlValue::new(wkt::Value::String(r#"[100, null, "world"]"#.to_string()));
        let parsed_json = GenericTupleSql::<i64, String>::from_value(json_val)
            .expect("should deserialize generic tuple struct from JSON array");
        assert_eq!(parsed_json, GenericTupleSql(100, None, "world".to_string()));
    }

    #[cfg(any(
        google_cloud_unstable_bigquery_arrow,
        google_cloud_unstable_bigquery_storage_read
    ))]
    mod arrow_tests {
        use super::*;
        use arrow::array::{
            ArrayRef, BinaryArray, BooleanArray, Date32Array, Decimal128Array, Decimal256Array,
            Float32Array, Float64Array, Int32Array, Int64Array, LargeBinaryArray, LargeListArray,
            LargeStringArray, ListArray, StringArray, StructArray, Time64MicrosecondArray,
            Time64NanosecondArray, TimestampMicrosecondArray, TimestampMillisecondArray,
            TimestampNanosecondArray, TimestampSecondArray,
        };
        use arrow::buffer::{NullBuffer, OffsetBuffer};
        use arrow::datatypes::{DataType, Field, Fields, i256};
        use std::sync::Arc;
        use test_case::test_case;

        fn arrow_val(arr: ArrayRef, row_idx: usize) -> SqlValue {
            SqlValue::from_inner(SqlValueInner::Arrow(ArrowCell::new(arr, row_idx)))
        }

        fn sample_list_i64(values: Vec<Option<i64>>, nulls: Option<NullBuffer>) -> ArrayRef {
            let len = values.len();
            let val_arr: ArrayRef = Arc::new(Int64Array::from(values));
            let offsets = OffsetBuffer::from_lengths(vec![len]);
            Arc::new(ListArray::new(
                Arc::new(Field::new("item", DataType::Int64, true)),
                offsets,
                val_arr,
                nulls,
            ))
        }

        fn sample_large_list_i64(values: Vec<Option<i64>>) -> ArrayRef {
            let len = values.len();
            let val_arr: ArrayRef = Arc::new(Int64Array::from(values));
            let offsets = OffsetBuffer::from_lengths(vec![len]);
            Arc::new(LargeListArray::new(
                Arc::new(Field::new("item", DataType::Int64, true)),
                offsets,
                val_arr,
                None,
            ))
        }

        fn decimal128_array(values: Vec<Option<i128>>, precision: u8, scale: i8) -> ArrayRef {
            Arc::new(
                Decimal128Array::from(values)
                    .with_precision_and_scale(precision, scale)
                    .unwrap(),
            )
        }

        fn decimal256_array(values: Vec<Option<i256>>, precision: u8, scale: i8) -> ArrayRef {
            Arc::new(
                Decimal256Array::from(values)
                    .with_precision_and_scale(precision, scale)
                    .unwrap(),
            )
        }

        #[test_case(Arc::new(BooleanArray::from(vec![true])) => "Boolean" ; "arrow boolean")]
        #[test_case(Arc::new(Int64Array::from(vec![42])) => "Int64" ; "arrow int64")]
        #[test_case(Arc::new(Float64Array::from(vec![3.25])) => "Float64" ; "arrow float64")]
        #[test_case(Arc::new(StringArray::from(vec!["hello"])) => "Utf8" ; "arrow utf8")]
        fn test_sql_value_inner_arrow_type_name(arr: ArrayRef) -> String {
            let val = SqlValueInner::Arrow(ArrowCell::new(arr, 0));
            let err = ConvertError::type_mismatch("expected_type", &val);
            assert!(matches!(
                err,
                ConvertError::TypeMismatch { ref expected, ref got }
                    if expected == "expected_type" && got == &val.type_name()
            ));
            val.type_name()
        }

        // TODO(#7032): Remove this test once wkt::Value conversion for Arrow cells is implemented.
        #[test]
        fn test_from_sql_value_arrow_not_implemented() {
            let arr: ArrayRef = Arc::new(Int64Array::from(vec![42]));
            let err = wkt::Value::from_value(arrow_val(arr, 0)).unwrap_err();
            assert!(matches!(err, ConvertError::Convert(_)));
            assert!(err.to_string().contains("not yet implemented"), "{err}");
        }

        #[test_case(Arc::new(BooleanArray::from(vec![Some(true)])) => Ok(true) ; "bool true")]
        #[test_case(Arc::new(BooleanArray::from(vec![Some(false)])) => Ok(false) ; "bool false")]
        #[test_case(Arc::new(BooleanArray::from(vec![None])) => Err(TestConvertError::NotNull) ; "bool null")]
        #[test_case(Arc::new(Int64Array::from(vec![1])) => Err(TestConvertError::type_mismatch("BooleanArray")) ; "bool type mismatch")]
        fn test_from_sql_arrow_bool(arr: ArrayRef) -> Result<bool, TestConvertError> {
            bool::from_value(arrow_val(arr, 0)).map_err(TestConvertError::from)
        }

        #[test_case(Arc::new(Int32Array::from(vec![Some(123)])) => Ok(123) ; "i32 from Int32Array")]
        #[test_case(Arc::new(Int64Array::from(vec![Some(456)])) => Ok(456) ; "i32 from Int64Array")]
        #[test_case(Arc::new(Int64Array::from(vec![Some(i64::from(i32::MAX) + 1)])) => Err(TestConvertError::Convert) ; "i32 overflow from Int64Array")]
        #[test_case(Arc::new(Int32Array::from(vec![None])) => Err(TestConvertError::NotNull) ; "i32 null")]
        #[test_case(Arc::new(BooleanArray::from(vec![true])) => Err(TestConvertError::type_mismatch("Int64Array or Int32Array")) ; "i32 type mismatch")]
        fn test_from_sql_arrow_i32(arr: ArrayRef) -> Result<i32, TestConvertError> {
            i32::from_value(arrow_val(arr, 0)).map_err(TestConvertError::from)
        }

        #[test_case(Arc::new(Int64Array::from(vec![Some(42)])) => Ok(42) ; "i64 valid")]
        #[test_case(Arc::new(Int64Array::from(vec![None])) => Err(TestConvertError::NotNull) ; "i64 null")]
        #[test_case(Arc::new(BooleanArray::from(vec![true])) => Err(TestConvertError::type_mismatch("Int64Array")) ; "i64 type mismatch")]
        fn test_from_sql_arrow_i64(arr: ArrayRef) -> Result<i64, TestConvertError> {
            i64::from_value(arrow_val(arr, 0)).map_err(TestConvertError::from)
        }

        #[test_case(Arc::new(Float32Array::from(vec![Some(1.5_f32)])) => Ok(1.5) ; "f32 from Float32Array")]
        #[test_case(Arc::new(Float64Array::from(vec![Some(2.5_f64)])) => Ok(2.5) ; "f32 from Float64Array")]
        #[test_case(Arc::new(Float32Array::from(vec![None])) => Err(TestConvertError::NotNull) ; "f32 null")]
        #[test_case(Arc::new(Int64Array::from(vec![1])) => Err(TestConvertError::type_mismatch("Float64Array or Float32Array")) ; "f32 type mismatch")]
        fn test_from_sql_arrow_f32(arr: ArrayRef) -> Result<f32, TestConvertError> {
            f32::from_value(arrow_val(arr, 0)).map_err(TestConvertError::from)
        }

        #[test_case(Arc::new(Float64Array::from(vec![Some(3.25)])) => Ok(3.25) ; "f64 valid")]
        #[test_case(Arc::new(Float64Array::from(vec![None])) => Err(TestConvertError::NotNull) ; "f64 null")]
        #[test_case(Arc::new(Int64Array::from(vec![1])) => Err(TestConvertError::type_mismatch("Float64Array")) ; "f64 type mismatch")]
        fn test_from_sql_arrow_f64(arr: ArrayRef) -> Result<f64, TestConvertError> {
            f64::from_value(arrow_val(arr, 0)).map_err(TestConvertError::from)
        }

        #[test_case(Arc::new(StringArray::from(vec![Some("hello")])) => Ok("hello".to_string()) ; "string from StringArray")]
        #[test_case(Arc::new(LargeStringArray::from(vec![Some("world")])) => Ok("world".to_string()) ; "string from LargeStringArray")]
        #[test_case(Arc::new(StringArray::from(vec![None::<&str>])) => Err(TestConvertError::NotNull) ; "string null")]
        #[test_case(Arc::new(Int64Array::from(vec![1])) => Err(TestConvertError::type_mismatch("StringArray or LargeStringArray")) ; "string type mismatch")]
        fn test_from_sql_arrow_string(arr: ArrayRef) -> Result<String, TestConvertError> {
            String::from_value(arrow_val(arr, 0)).map_err(TestConvertError::from)
        }

        #[test_case(Arc::new(BinaryArray::from(vec![Some(b"abc".as_slice())])) => Ok(b"abc".to_vec()) ; "vec u8 from BinaryArray")]
        #[test_case(Arc::new(LargeBinaryArray::from(vec![Some(b"xyz".as_slice())])) => Ok(b"xyz".to_vec()) ; "vec u8 from LargeBinaryArray")]
        #[test_case(Arc::new(BinaryArray::from(vec![None::<&[u8]>])) => Err(TestConvertError::NotNull) ; "vec u8 null")]
        #[test_case(Arc::new(Int64Array::from(vec![1])) => Err(TestConvertError::type_mismatch("BinaryArray or LargeBinaryArray")) ; "vec u8 type mismatch")]
        fn test_from_sql_arrow_vec_u8(arr: ArrayRef) -> Result<Vec<u8>, TestConvertError> {
            Vec::<u8>::from_value(arrow_val(arr, 0)).map_err(TestConvertError::from)
        }

        #[test_case(Arc::new(BinaryArray::from(vec![Some(b"abc".as_slice())])) => Ok(bytes::Bytes::from_static(b"abc")) ; "bytes from BinaryArray")]
        #[test_case(Arc::new(LargeBinaryArray::from(vec![Some(b"xyz".as_slice())])) => Ok(bytes::Bytes::from_static(b"xyz")) ; "bytes from LargeBinaryArray")]
        #[test_case(Arc::new(BinaryArray::from(vec![None::<&[u8]>])) => Err(TestConvertError::NotNull) ; "bytes null")]
        #[test_case(Arc::new(Int64Array::from(vec![1])) => Err(TestConvertError::type_mismatch("BinaryArray or LargeBinaryArray")) ; "bytes type mismatch")]
        fn test_from_sql_arrow_bytes(arr: ArrayRef) -> Result<bytes::Bytes, TestConvertError> {
            bytes::Bytes::from_value(arrow_val(arr, 0)).map_err(TestConvertError::from)
        }

        #[test_case(decimal128_array(vec![Some(123_456)], 10, 3) => Ok(Decimal::new().set_value("123.456")) ; "decimal from Decimal128Array")]
        #[test_case(decimal128_array(vec![Some(-123_456)], 10, 3) => Ok(Decimal::new().set_value("-123.456")) ; "decimal negative from Decimal128Array")]
        #[test_case(decimal256_array(vec![Some(i256::from_i128(789_012))], 20, 3) => Ok(Decimal::new().set_value("789.012")) ; "decimal from Decimal256Array")]
        #[test_case(decimal128_array(vec![None], 10, 3) => Err(TestConvertError::NotNull) ; "decimal null Decimal128Array")]
        #[test_case(decimal256_array(vec![None], 20, 3) => Err(TestConvertError::NotNull) ; "decimal null Decimal256Array")]
        #[test_case(Arc::new(Int64Array::from(vec![1])) => Err(TestConvertError::type_mismatch("Decimal128Array or Decimal256Array")) ; "decimal type mismatch")]
        fn test_from_sql_arrow_decimal(arr: ArrayRef) -> Result<Decimal, TestConvertError> {
            Decimal::from_value(arrow_val(arr, 0)).map_err(TestConvertError::from)
        }

        #[test_case(decimal128_array(vec![Some(123_456)], 10, 3) => Ok(RustDecimal::from_str_exact("123.456").unwrap()) ; "rust_decimal from Decimal128Array")]
        #[test_case(decimal128_array(vec![Some(-123_456)], 10, 3) => Ok(RustDecimal::from_str_exact("-123.456").unwrap()) ; "rust_decimal negative from Decimal128Array")]
        #[test_case(decimal128_array(vec![Some(100_000_000_000_000_000_000_000_000_000_i128)], 38, 9) => Ok(RustDecimal::from_str_exact("100000000000000000000").unwrap()) ; "rust_decimal from Decimal128Array with trailing zeros exceeding 96 bits unscaled")]
        #[test_case(decimal128_array(vec![Some(12)], 10, -2) => Ok(RustDecimal::from(1200)) ; "rust_decimal from Decimal128Array with negative scale")]
        #[test_case(decimal128_array(vec![Some(i128::MAX)], 38, 9) => Err(TestConvertError::Convert) ; "rust_decimal overflow from Decimal128Array")]
        #[test_case(decimal256_array(vec![Some(i256::from_string("123456789012345678900000000000000000000000000000").unwrap())], 76, 38) => Ok(RustDecimal::from_str_exact("1234567890.123456789").unwrap()) ; "rust_decimal from Decimal256Array with trailing zeros")]
        #[test_case(decimal256_array(vec![Some(i256::from_string("4200000000000000000000000000000000000000").unwrap())], 76, 38) => Ok(RustDecimal::from(42)) ; "rust_decimal from Decimal256Array integer with trailing zeros")]
        #[test_case(decimal256_array(vec![Some(i256::from_i128(12))], 20, -2) => Ok(RustDecimal::from(1200)) ; "rust_decimal from Decimal256Array without decimal point")]
        #[test_case(decimal256_array(vec![Some(i256::MAX)], 76, 38) => Err(TestConvertError::Convert) ; "rust_decimal overflow from Decimal256Array")]
        #[test_case(decimal128_array(vec![None], 10, 3) => Err(TestConvertError::NotNull) ; "rust_decimal null Decimal128Array")]
        #[test_case(decimal256_array(vec![None], 20, 3) => Err(TestConvertError::NotNull) ; "rust_decimal null Decimal256Array")]
        #[test_case(Arc::new(Int64Array::from(vec![1])) => Err(TestConvertError::type_mismatch("Decimal128Array or Decimal256Array")) ; "rust_decimal type mismatch")]
        fn test_from_sql_arrow_rust_decimal(
            arr: ArrayRef,
        ) -> Result<RustDecimal, TestConvertError> {
            RustDecimal::from_value(arrow_val(arr, 0)).map_err(TestConvertError::from)
        }

        #[test_case(
            decimal128_array(vec![Some(42_500_000_000)], 38, 9)
            => ("42.5".to_string(), 1)
            ; "Decimal128 fractional trailing zeros normalized"
        )]
        #[test_case(
            decimal128_array(vec![Some(42_000_000_000)], 38, 9)
            => ("42".to_string(), 0)
            ; "Decimal128 integer trailing zeros normalized"
        )]
        #[test_case(
            decimal256_array(vec![Some(i256::from_string("4250000000000000000000000000000000000000").unwrap())], 76, 38)
            => ("42.5".to_string(), 1)
            ; "Decimal256 fractional trailing zeros normalized"
        )]
        fn test_from_sql_arrow_rust_decimal_scale(arr: ArrayRef) -> (String, u32) {
            let d = RustDecimal::from_value(arrow_val(arr, 0)).unwrap();
            (d.to_string(), d.scale())
        }

        #[test_case(Arc::new(Int64Array::from(vec![Some(42)])) => Ok(Some(42)) ; "option some i64")]
        #[test_case(Arc::new(Int64Array::from(vec![None])) => Ok(None) ; "option null")]
        #[test_case(Arc::new(BooleanArray::from(vec![Some(true)])) => Err(TestConvertError::type_mismatch("Int64Array")) ; "option type mismatch")]
        fn test_from_sql_arrow_option(arr: ArrayRef) -> Result<Option<i64>, TestConvertError> {
            Option::<i64>::from_value(arrow_val(arr, 0)).map_err(TestConvertError::from)
        }

        #[test_case(Arc::new(TimestampMicrosecondArray::from(vec![Some(1_779_982_200_123_456)])) => Ok(wkt::Timestamp::new(1_779_982_200, 123_456_000).unwrap()) ; "timestamp microsecond")]
        #[test_case(Arc::new(TimestampMillisecondArray::from(vec![Some(1_779_982_200_123)])) => Ok(wkt::Timestamp::new(1_779_982_200, 123_000_000).unwrap()) ; "timestamp millisecond")]
        #[test_case(Arc::new(TimestampNanosecondArray::from(vec![Some(1_779_982_200_123_456_789)])) => Ok(wkt::Timestamp::new(1_779_982_200, 123_456_000).unwrap()) ; "timestamp nanosecond")]
        #[test_case(Arc::new(TimestampSecondArray::from(vec![Some(1_779_982_200)])) => Ok(wkt::Timestamp::new(1_779_982_200, 0).unwrap()) ; "timestamp second")]
        #[test_case(Arc::new(TimestampMicrosecondArray::from(vec![Some(-1)])) => Ok(wkt::Timestamp::new(-1, 999_999_000).unwrap()) ; "timestamp pre epoch")]
        #[test_case(Arc::new(TimestampMicrosecondArray::from(vec![None])) => Err(TestConvertError::NotNull) ; "timestamp null")]
        #[test_case(Arc::new(Int64Array::from(vec![1])) => Err(TestConvertError::type_mismatch("TimestampArray")) ; "timestamp type mismatch")]
        fn test_from_sql_arrow_timestamp(
            arr: ArrayRef,
        ) -> Result<wkt::Timestamp, TestConvertError> {
            wkt::Timestamp::from_value(arrow_val(arr, 0)).map_err(TestConvertError::from)
        }

        #[test_case(Arc::new(Date32Array::from(vec![Some(20_601)])) => Ok(google_cloud_type::model::Date::new().set_year(2026).set_month(5).set_day(28)) ; "date valid")]
        #[test_case(Arc::new(Date32Array::from(vec![Some(0)])) => Ok(google_cloud_type::model::Date::new().set_year(1970).set_month(1).set_day(1)) ; "date epoch")]
        #[test_case(Arc::new(Date32Array::from(vec![Some(i32::MAX)])) => Err(TestConvertError::Convert) ; "date out of range")]
        #[test_case(Arc::new(Date32Array::from(vec![None])) => Err(TestConvertError::NotNull) ; "date null")]
        #[test_case(Arc::new(Int32Array::from(vec![20_601])) => Err(TestConvertError::type_mismatch("Date32Array")) ; "date type mismatch")]
        fn test_from_sql_arrow_date(
            arr: ArrayRef,
        ) -> Result<google_cloud_type::model::Date, TestConvertError> {
            google_cloud_type::model::Date::from_value(arrow_val(arr, 0))
                .map_err(TestConvertError::from)
        }

        #[test_case(Arc::new(Time64MicrosecondArray::from(vec![Some(55_800_000_000)])) => Ok(google_cloud_type::model::TimeOfDay::new().set_hours(15).set_minutes(30).set_seconds(0).set_nanos(0)) ; "time of day valid")]
        #[test_case(Arc::new(Time64MicrosecondArray::from(vec![Some(55_800_123_456)])) => Ok(google_cloud_type::model::TimeOfDay::new().set_hours(15).set_minutes(30).set_seconds(0).set_nanos(123_456_000)) ; "time of day fractional")]
        #[test_case(Arc::new(Time64NanosecondArray::from(vec![Some(55_800_123_456_789)])) => Ok(google_cloud_type::model::TimeOfDay::new().set_hours(15).set_minutes(30).set_seconds(0).set_nanos(123_456_000)) ; "time of day nanosecond")]
        #[test_case(Arc::new(Time64MicrosecondArray::from(vec![Some(-1)])) => Err(TestConvertError::Convert) ; "time of day negative")]
        #[test_case(Arc::new(Time64MicrosecondArray::from(vec![Some(86_400_000_000)])) => Err(TestConvertError::Convert) ; "time of day 24h out of range")]
        #[test_case(Arc::new(Time64MicrosecondArray::from(vec![None])) => Err(TestConvertError::NotNull) ; "time of day null")]
        #[test_case(Arc::new(Int64Array::from(vec![1])) => Err(TestConvertError::type_mismatch("Time64MicrosecondArray")) ; "time of day type mismatch")]
        fn test_from_sql_arrow_time_of_day(
            arr: ArrayRef,
        ) -> Result<google_cloud_type::model::TimeOfDay, TestConvertError> {
            google_cloud_type::model::TimeOfDay::from_value(arrow_val(arr, 0))
                .map_err(TestConvertError::from)
        }

        #[test_case(Arc::new(TimestampMicrosecondArray::from(vec![Some(1_779_982_200_000_000)])) => Ok(google_cloud_type::model::DateTime::new().set_year(2026).set_month(5).set_day(28).set_hours(15).set_minutes(30).set_seconds(0).set_nanos(0)) ; "datetime without subseconds")]
        #[test_case(Arc::new(TimestampMicrosecondArray::from(vec![Some(1_779_982_200_123_456)])) => Ok(google_cloud_type::model::DateTime::new().set_year(2026).set_month(5).set_day(28).set_hours(15).set_minutes(30).set_seconds(0).set_nanos(123_456_000)) ; "datetime with subseconds")]
        #[test_case(Arc::new(TimestampMicrosecondArray::from(vec![Some(i64::MAX)])) => Err(TestConvertError::Convert) ; "datetime out of range")]
        #[test_case(Arc::new(TimestampMicrosecondArray::from(vec![None])) => Err(TestConvertError::NotNull) ; "datetime null")]
        #[test_case(Arc::new(Int64Array::from(vec![1])) => Err(TestConvertError::type_mismatch("TimestampArray")) ; "datetime type mismatch")]
        fn test_from_sql_arrow_datetime(
            arr: ArrayRef,
        ) -> Result<google_cloud_type::model::DateTime, TestConvertError> {
            google_cloud_type::model::DateTime::from_value(arrow_val(arr, 0))
                .map_err(TestConvertError::from)
        }

        #[test_case(sample_list_i64(vec![Some(10), Some(20), Some(30)], None) => Ok(vec![10, 20, 30]) ; "vec from ListArray")]
        #[test_case(sample_large_list_i64(vec![Some(40), Some(50)]) => Ok(vec![40, 50]) ; "vec from LargeListArray")]
        #[test_case(sample_list_i64(vec![], None) => Ok(vec![]) ; "vec empty ListArray")]
        #[test_case(sample_list_i64(vec![Some(10)], Some(NullBuffer::from(vec![false]))) => Err(TestConvertError::NotNull) ; "vec null ListArray")]
        #[test_case(sample_list_i64(vec![Some(10), None], None) => Err(TestConvertError::NotNull) ; "vec null element in non-optional Vec")]
        #[test_case(Arc::new(Int64Array::from(vec![1])) => Err(TestConvertError::type_mismatch("ListArray or LargeListArray")) ; "vec type mismatch")]
        fn test_from_sql_arrow_vec_i64(arr: ArrayRef) -> Result<Vec<i64>, TestConvertError> {
            Vec::<i64>::from_value(arrow_val(arr, 0)).map_err(TestConvertError::from)
        }

        #[test_case(sample_list_i64(vec![Some(10), None, Some(30)], None) => Ok(vec![Some(10), None, Some(30)]) ; "vec option i64 with null element")]
        fn test_from_sql_arrow_vec_option_i64(
            arr: ArrayRef,
        ) -> Result<Vec<Option<i64>>, TestConvertError> {
            Vec::<Option<i64>>::from_value(arrow_val(arr, 0)).map_err(TestConvertError::from)
        }

        #[test]
        fn test_sql_value_take_arrow_list_by_index() -> anyhow::Result<()> {
            let mut list_val = arrow_val(sample_list_i64(vec![Some(10), Some(20)], None), 0);
            assert_eq!(list_val.take::<i64, _>(0)?, 10);
            assert_eq!(list_val.take::<i64, _>(1)?, 20);
            assert!(matches!(
                list_val.take::<i64, _>(2),
                Err(ConvertError::MissingField(ref f)) if f == "2"
            ));

            Ok(())
        }

        fn sample_test_struct_array(nulls: Option<NullBuffer>) -> ArrayRef {
            let fields = Fields::from(vec![
                Field::new("name", DataType::Utf8, false),
                Field::new("custom_int", DataType::Int64, false),
                Field::new("some_bool", DataType::Boolean, false),
            ]);
            Arc::new(StructArray::new(
                fields,
                vec![
                    Arc::new(StringArray::from(vec!["James"])) as ArrayRef,
                    Arc::new(Int64Array::from(vec![272_793])) as ArrayRef,
                    Arc::new(BooleanArray::from(vec![true])) as ArrayRef,
                ],
                nulls,
            ))
        }

        fn sample_missing_field_struct_array() -> ArrayRef {
            let fields = Fields::from(vec![
                Field::new("name", DataType::Utf8, false),
                Field::new("some_bool", DataType::Boolean, false),
            ]);
            Arc::new(StructArray::new(
                fields,
                vec![
                    Arc::new(StringArray::from(vec!["James"])) as ArrayRef,
                    Arc::new(BooleanArray::from(vec![true])) as ArrayRef,
                ],
                None,
            ))
        }

        fn sample_tuple_struct_array(nulls: Option<NullBuffer>) -> ArrayRef {
            let fields = Fields::from(vec![
                Field::new("", DataType::Int64, false),
                Field::new("", DataType::Int64, true),
                Field::new("", DataType::Utf8, false),
            ]);
            Arc::new(StructArray::new(
                fields,
                vec![
                    Arc::new(Int64Array::from(vec![Some(100)])) as ArrayRef,
                    Arc::new(Int64Array::from(vec![None])) as ArrayRef,
                    Arc::new(StringArray::from(vec!["world"])) as ArrayRef,
                ],
                nulls,
            ))
        }

        fn sample_short_tuple_struct_array() -> ArrayRef {
            let fields = Fields::from(vec![
                Field::new("", DataType::Int64, false),
                Field::new("", DataType::Int64, true),
            ]);
            Arc::new(StructArray::new(
                fields,
                vec![
                    Arc::new(Int64Array::from(vec![Some(100)])) as ArrayRef,
                    Arc::new(Int64Array::from(vec![None])) as ArrayRef,
                ],
                None,
            ))
        }

        #[test_case(sample_test_struct_array(None) => Ok(TestSqlStruct { name: "James".to_string(), some_int: 272_793, some_bool: true }) ; "struct from StructArray")]
        #[test_case(Arc::new(StringArray::from(vec![r#"{"name": "James", "custom_int": 272793, "some_bool": true}"#])) => Ok(TestSqlStruct { name: "James".to_string(), some_int: 272_793, some_bool: true }) ; "struct from StringArray json")]
        #[test_case(Arc::new(LargeStringArray::from(vec![r#"{"name": "James", "custom_int": 272793, "some_bool": true}"#])) => Ok(TestSqlStruct { name: "James".to_string(), some_int: 272_793, some_bool: true }) ; "struct from LargeStringArray json")]
        #[test_case(sample_missing_field_struct_array() => Err(TestConvertError::MissingField("custom_int".to_string())) ; "struct missing field")]
        #[test_case(sample_test_struct_array(Some(NullBuffer::from(vec![false]))) => Err(TestConvertError::NotNull) ; "struct null StructArray")]
        #[test_case(Arc::new(StringArray::from(vec!["invalid"])) => Err(TestConvertError::Convert) ; "struct invalid json string")]
        #[test_case(Arc::new(BooleanArray::from(vec![true])) => Err(TestConvertError::type_mismatch("object or string")) ; "struct type mismatch")]
        fn test_derive_from_sql_arrow_named_struct(
            arr: ArrayRef,
        ) -> Result<TestSqlStruct, TestConvertError> {
            TestSqlStruct::from_value(arrow_val(arr, 0)).map_err(TestConvertError::from)
        }

        #[test_case(sample_tuple_struct_array(None) => Ok(GenericTupleSql(100, None, "world".to_string())) ; "tuple struct from StructArray")]
        #[test_case(Arc::new(StringArray::from(vec![r#"[100, null, "world"]"#])) => Ok(GenericTupleSql(100, None, "world".to_string())) ; "tuple struct from StringArray json")]
        #[test_case(sample_short_tuple_struct_array() => Err(TestConvertError::MissingField("2".to_string())) ; "tuple struct missing field")]
        #[test_case(sample_tuple_struct_array(Some(NullBuffer::from(vec![false]))) => Err(TestConvertError::NotNull) ; "tuple struct null StructArray")]
        #[test_case(Arc::new(BooleanArray::from(vec![true])) => Err(TestConvertError::type_mismatch("struct, array, or string")) ; "tuple struct type mismatch")]
        fn test_derive_from_sql_arrow_tuple_struct(
            arr: ArrayRef,
        ) -> Result<GenericTupleSql<i64, String>, TestConvertError> {
            GenericTupleSql::<i64, String>::from_value(arrow_val(arr, 0))
                .map_err(TestConvertError::from)
        }

        #[test_case(Arc::new(StringArray::from(vec![Some(r#"{"a": 1}"#)])) => Ok(wkt::Struct::from_iter([("a".to_string(), wkt::Value::Number(1.into()))])) ; "wkt struct from StringArray")]
        #[test_case(Arc::new(LargeStringArray::from(vec![Some(r#"{"a": 1}"#)])) => Ok(wkt::Struct::from_iter([("a".to_string(), wkt::Value::Number(1.into()))])) ; "wkt struct from LargeStringArray")]
        #[test_case(Arc::new(StringArray::from(vec![None::<&str>])) => Err(TestConvertError::NotNull) ; "wkt struct null")]
        #[test_case(Arc::new(StringArray::from(vec![Some("invalid")])) => Err(TestConvertError::Convert) ; "wkt struct invalid json")]
        #[test_case(Arc::new(BooleanArray::from(vec![true])) => Err(TestConvertError::type_mismatch("StringArray or LargeStringArray")) ; "wkt struct type mismatch")]
        fn test_from_sql_arrow_wkt_struct(arr: ArrayRef) -> Result<wkt::Struct, TestConvertError> {
            wkt::Struct::from_value(arrow_val(arr, 0)).map_err(TestConvertError::from)
        }
    }
}