mssql-client 0.13.0

//! Column value parsing for TDS row data.
//!
//! This module contains all logic for parsing SQL Server column values from
//! raw TDS wire bytes into `SqlValue` types. It handles:
//!
//! - Fixed-length types (INT, BIGINT, FLOAT, BIT, etc.)
//! - Variable-length nullable types (IntN, FloatN, BitN, etc.)
//! - Date/time types (DATE, TIME, DATETIME, DATETIME2, DATETIMEOFFSET)
//! - String types (VARCHAR, NVARCHAR, CHAR, NCHAR, TEXT, NTEXT)
//! - Binary types (VARBINARY, BINARY, IMAGE)
//! - Decimal/Numeric types
//! - Money types (MONEY, SMALLMONEY)
//! - PLP (Partially Length-Prefixed) encoding for MAX types
//! - SQL_VARIANT with embedded type information
//! - XML, GUID, UDT
//!
//! All functions are pure (no `self` parameter) and operate on byte buffers
//! with TDS column metadata.

// Allow unwrap/expect ONLY for chrono construction from compile-time
// constants (epochs, midnight, UTC offset 0). Values that arrive from the
// wire must use checked construction and surface protocol errors — a
// malicious or buggy server must never be able to panic the client (see
// `smalldatetime_from_wire` / `datetime_from_wire`).
#![allow(clippy::unwrap_used, clippy::expect_used, clippy::needless_range_loop)]

use bytes::Buf;
use mssql_types::SqlValue;
use tds_protocol::token::{ColMetaData, Collation, ColumnData, NbcRow, RawRow};
use tds_protocol::types::TypeId;

use crate::error::{Error, Result};

/// Build a `NaiveDateTime` from SMALLDATETIME wire values (days since
/// 1900-01-01, minutes since midnight). Both come from the wire: out-of-range
/// values are protocol errors, never panics.
#[cfg(feature = "chrono")]
fn smalldatetime_from_wire(days: i64, minutes: u32) -> Result<chrono::NaiveDateTime> {
    let base = chrono::NaiveDate::from_ymd_opt(1900, 1, 1).expect("epoch 1900-01-01 is valid");
    let date = base
        .checked_add_signed(chrono::Duration::days(days))
        .ok_or_else(|| Error::Protocol(format!("SMALLDATETIME days out of range: {days}")))?;
    let secs = u64::from(minutes) * 60;
    let time = u32::try_from(secs)
        .ok()
        .and_then(|s| chrono::NaiveTime::from_num_seconds_from_midnight_opt(s, 0))
        .ok_or_else(|| Error::Protocol(format!("SMALLDATETIME minutes out of range: {minutes}")))?;
    Ok(date.and_time(time))
}

/// Build a `NaiveDateTime` from DATETIME wire values (days since 1900-01-01,
/// 1/300ths of a second since midnight). Both come from the wire:
/// out-of-range values are protocol errors, never panics.
#[cfg(feature = "chrono")]
fn datetime_from_wire(days: i64, time_300ths: u64) -> Result<chrono::NaiveDateTime> {
    let base = chrono::NaiveDate::from_ymd_opt(1900, 1, 1).expect("epoch 1900-01-01 is valid");
    let date = base
        .checked_add_signed(chrono::Duration::days(days))
        .ok_or_else(|| Error::Protocol(format!("DATETIME days out of range: {days}")))?;
    let total_ms = (time_300ths * 1000) / 300;
    let nanos = ((total_ms % 1000) * 1_000_000) as u32;
    let time = u32::try_from(total_ms / 1000)
        .ok()
        .and_then(|secs| chrono::NaiveTime::from_num_seconds_from_midnight_opt(secs, nanos))
        .ok_or_else(|| {
            Error::Protocol(format!(
                "DATETIME time component out of range: {time_300ths}"
            ))
        })?;
    Ok(date.and_time(time))
}

/// Convert a RawRow to a client Row.
///
/// This parses the raw bytes back into SqlValue types based on column metadata.
pub(crate) fn convert_raw_row(
    raw: &RawRow,
    meta: &ColMetaData,
    columns: &[crate::row::Column],
) -> Result<crate::row::Row> {
    let mut values = Vec::with_capacity(meta.columns.len());
    let mut buf = raw.data.as_ref();

    for col in &meta.columns {
        let value = parse_column_value(&mut buf, col)?;
        values.push(value);
    }

    Ok(crate::row::Row::from_values(columns.to_vec(), values))
}

/// Convert an NbcRow to a client Row.
///
/// NbcRow has a null bitmap followed by only non-null values.
pub(crate) fn convert_nbc_row(
    nbc: &NbcRow,
    meta: &ColMetaData,
    columns: &[crate::row::Column],
) -> Result<crate::row::Row> {
    let mut values = Vec::with_capacity(meta.columns.len());
    let mut buf = nbc.data.as_ref();

    for (i, col) in meta.columns.iter().enumerate() {
        if nbc.is_null(i) {
            values.push(mssql_types::SqlValue::Null);
        } else {
            let value = parse_column_value(&mut buf, col)?;
            values.push(value);
        }
    }

    Ok(crate::row::Row::from_values(columns.to_vec(), values))
}

/// Convert a RawRow to a client Row with Always Encrypted decryption.
///
/// For encrypted columns (identified by the decryptor), this:
/// 1. Parses the column value as raw bytes (BigVarBinary on the wire)
/// 2. Decrypts the ciphertext using the pre-resolved encryptor
/// 3. Re-parses the decrypted plaintext using the base column type
///
/// For unencrypted columns, this delegates to `parse_column_value` as normal.
#[cfg(feature = "always-encrypted")]
pub(crate) fn convert_raw_row_decrypted(
    raw: &RawRow,
    meta: &ColMetaData,
    columns: &[crate::row::Column],
    decryptor: &crate::column_decryptor::ColumnDecryptor,
) -> Result<crate::row::Row> {
    let mut values = Vec::with_capacity(meta.columns.len());
    let mut buf = raw.data.as_ref();

    for (i, col) in meta.columns.iter().enumerate() {
        let value = if decryptor.is_encrypted(i) {
            decrypt_column(&mut buf, col, decryptor, i)?
        } else {
            parse_column_value(&mut buf, col)?
        };
        values.push(value);
    }

    Ok(crate::row::Row::from_values(columns.to_vec(), values))
}

/// Convert an NbcRow to a client Row with Always Encrypted decryption.
///
/// Same as `convert_raw_row_decrypted` but handles the null bitmap.
#[cfg(feature = "always-encrypted")]
pub(crate) fn convert_nbc_row_decrypted(
    nbc: &NbcRow,
    meta: &ColMetaData,
    columns: &[crate::row::Column],
    decryptor: &crate::column_decryptor::ColumnDecryptor,
) -> Result<crate::row::Row> {
    let mut values = Vec::with_capacity(meta.columns.len());
    let mut buf = nbc.data.as_ref();

    for (i, col) in meta.columns.iter().enumerate() {
        if nbc.is_null(i) {
            values.push(SqlValue::Null);
        } else {
            let value = if decryptor.is_encrypted(i) {
                decrypt_column(&mut buf, col, decryptor, i)?
            } else {
                parse_column_value(&mut buf, col)?
            };
            values.push(value);
        }
    }

    Ok(crate::row::Row::from_values(columns.to_vec(), values))
}

/// Decrypt an encrypted column value and re-parse it as the plaintext type.
///
/// Encrypted columns are transmitted as BigVarBinary (2-byte length prefix).
/// This function reads the ciphertext, decrypts it, then re-parses the
/// plaintext bytes using the base column type from CryptoMetadata.
#[cfg(feature = "always-encrypted")]
fn decrypt_column(
    buf: &mut &[u8],
    _col: &ColumnData,
    decryptor: &crate::column_decryptor::ColumnDecryptor,
    ordinal: usize,
) -> Result<SqlValue> {
    // Encrypted column wire type is BigVarBinary: 2-byte length prefix.
    // 0xFFFF = NULL, otherwise length followed by ciphertext bytes.
    if buf.remaining() < 2 {
        return Err(Error::Protocol(
            "unexpected EOF reading encrypted column length".to_string(),
        ));
    }

    let length = buf.get_u16_le();

    if length == 0xFFFF {
        // NULL encrypted value
        return Ok(SqlValue::Null);
    }

    let length = length as usize;
    if buf.remaining() < length {
        return Err(Error::Protocol(format!(
            "unexpected EOF reading encrypted column data: need {length} bytes, have {}",
            buf.remaining()
        )));
    }

    // Extract ciphertext bytes
    let ciphertext = &buf[..length];
    buf.advance(length);

    // Decrypt and get the base column metadata
    let (plaintext, base_col) = decryptor.decrypt_column_value(ordinal, ciphertext)?;

    // Re-parse the decrypted plaintext using the base column's type info
    let mut pt_buf: &[u8] = &plaintext;
    parse_column_value(&mut pt_buf, base_col)
}

/// Parse money value from buffer and convert to appropriate type.
///
/// Money is stored as fixed-point with 4 decimal places.
/// - 4 bytes: SMALLMONEY
/// - 8 bytes: MONEY
fn parse_money_value(buf: &mut &[u8], bytes: usize) -> Result<SqlValue> {
    if bytes == 0 {
        return Ok(SqlValue::Null);
    }

    let cents = match bytes {
        4 => buf.get_i32_le() as i64,
        8 => {
            let high = buf.get_i32_le();
            let low = buf.get_u32_le();
            ((high as i64) << 32) | (low as i64)
        }
        _ => return Err(Error::Protocol(format!("invalid money length: {bytes}"))),
    };

    #[cfg(feature = "decimal")]
    {
        use rust_decimal::Decimal;
        // `cents` is an i64 (fixed 4- or 8-byte MONEY) at scale 4, always
        // inside rust_decimal's 96-bit range — but use the checked
        // constructor anyway so no panicking `from_i128_with_scale` call
        // remains in the decoder (invariant: grep returns nothing).
        match Decimal::try_from_i128_with_scale(cents as i128, 4) {
            Ok(decimal) => Ok(SqlValue::Decimal(decimal)),
            Err(_) => Ok(SqlValue::Double((cents as f64) / 10000.0)),
        }
    }

    #[cfg(not(feature = "decimal"))]
    {
        Ok(SqlValue::Double((cents as f64) / 10000.0))
    }
}

/// Parse a single column value from a buffer based on column metadata.
// `pub` for the `__fuzzing` re-export; the module itself is `pub(crate)`,
// so this stays crate-private unless the `fuzzing` feature is enabled.
pub fn parse_column_value(buf: &mut &[u8], col: &ColumnData) -> Result<SqlValue> {
    let value = match col.type_id {
        // Fixed-length null type
        TypeId::Null => SqlValue::Null,

        // 1-byte types
        TypeId::Int1 => {
            if buf.remaining() < 1 {
                return Err(Error::Protocol("unexpected EOF reading TINYINT".into()));
            }
            SqlValue::TinyInt(buf.get_u8())
        }
        TypeId::Bit => {
            if buf.remaining() < 1 {
                return Err(Error::Protocol("unexpected EOF reading BIT".into()));
            }
            SqlValue::Bool(buf.get_u8() != 0)
        }

        // 2-byte types
        TypeId::Int2 => {
            if buf.remaining() < 2 {
                return Err(Error::Protocol("unexpected EOF reading SMALLINT".into()));
            }
            SqlValue::SmallInt(buf.get_i16_le())
        }

        // 4-byte types
        TypeId::Int4 => {
            if buf.remaining() < 4 {
                return Err(Error::Protocol("unexpected EOF reading INT".into()));
            }
            SqlValue::Int(buf.get_i32_le())
        }
        TypeId::Float4 => {
            if buf.remaining() < 4 {
                return Err(Error::Protocol("unexpected EOF reading REAL".into()));
            }
            SqlValue::Float(buf.get_f32_le())
        }

        // 8-byte types
        TypeId::Int8 => {
            if buf.remaining() < 8 {
                return Err(Error::Protocol("unexpected EOF reading BIGINT".into()));
            }
            SqlValue::BigInt(buf.get_i64_le())
        }
        TypeId::Float8 => {
            if buf.remaining() < 8 {
                return Err(Error::Protocol("unexpected EOF reading FLOAT".into()));
            }
            SqlValue::Double(buf.get_f64_le())
        }

        // Money types (fixed-point with 4 decimal places)
        TypeId::Money | TypeId::Money4 | TypeId::MoneyN => {
            let bytes = match col.type_id {
                TypeId::Money => 8,
                TypeId::Money4 => 4,
                TypeId::MoneyN => {
                    if buf.remaining() < 1 {
                        return Err(Error::Protocol(
                            "unexpected EOF reading MoneyN length".into(),
                        ));
                    }
                    buf.get_u8() as usize
                }
                // The outer match arm restricts col.type_id to Money | Money4 | MoneyN,
                // so this branch is unreachable.
                _ => unreachable!("inner match is bounded by outer Money|Money4|MoneyN arm"),
            };

            if buf.remaining() < bytes {
                return Err(Error::Protocol(format!(
                    "unexpected EOF reading money data ({bytes} bytes)"
                )));
            }

            parse_money_value(buf, bytes)?
        }

        // Variable-length nullable types (IntN, FloatN, etc.)
        TypeId::IntN => {
            if buf.remaining() < 1 {
                return Err(Error::Protocol("unexpected EOF reading IntN length".into()));
            }
            let len = buf.get_u8();
            if buf.remaining() < len as usize {
                return Err(Error::Protocol("unexpected EOF reading IntN data".into()));
            }
            match len {
                0 => SqlValue::Null,
                1 => SqlValue::TinyInt(buf.get_u8()),
                2 => SqlValue::SmallInt(buf.get_i16_le()),
                4 => SqlValue::Int(buf.get_i32_le()),
                8 => SqlValue::BigInt(buf.get_i64_le()),
                _ => {
                    return Err(Error::Protocol(format!("invalid IntN length: {len}")));
                }
            }
        }
        TypeId::FloatN => {
            if buf.remaining() < 1 {
                return Err(Error::Protocol(
                    "unexpected EOF reading FloatN length".into(),
                ));
            }
            let len = buf.get_u8();
            if buf.remaining() < len as usize {
                return Err(Error::Protocol("unexpected EOF reading FloatN data".into()));
            }
            match len {
                0 => SqlValue::Null,
                4 => SqlValue::Float(buf.get_f32_le()),
                8 => SqlValue::Double(buf.get_f64_le()),
                _ => {
                    return Err(Error::Protocol(format!("invalid FloatN length: {len}")));
                }
            }
        }
        TypeId::BitN => {
            if buf.remaining() < 1 {
                return Err(Error::Protocol("unexpected EOF reading BitN length".into()));
            }
            let len = buf.get_u8();
            if buf.remaining() < len as usize {
                return Err(Error::Protocol("unexpected EOF reading BitN data".into()));
            }
            match len {
                0 => SqlValue::Null,
                1 => SqlValue::Bool(buf.get_u8() != 0),
                _ => {
                    return Err(Error::Protocol(format!("invalid BitN length: {len}")));
                }
            }
        }

        // DECIMAL/NUMERIC types (1-byte length prefix)
        TypeId::Decimal | TypeId::Numeric | TypeId::DecimalN | TypeId::NumericN => {
            if buf.remaining() < 1 {
                return Err(Error::Protocol(
                    "unexpected EOF reading DECIMAL/NUMERIC length".into(),
                ));
            }
            let len = buf.get_u8() as usize;
            if len == 0 {
                SqlValue::Null
            } else {
                if buf.remaining() < len {
                    return Err(Error::Protocol(
                        "unexpected EOF reading DECIMAL/NUMERIC data".into(),
                    ));
                }

                // First byte is sign: 0 = negative, 1 = positive
                let sign = buf.get_u8();
                let mantissa_len = len - 1;

                // Read mantissa as little-endian integer (up to 16 bytes for max precision 38)
                let mut mantissa_bytes = [0u8; 16];
                for i in 0..mantissa_len.min(16) {
                    mantissa_bytes[i] = buf.get_u8();
                }
                // Skip any excess bytes (shouldn't happen with valid data)
                for _ in 16..mantissa_len {
                    buf.get_u8();
                }

                let mantissa = u128::from_le_bytes(mantissa_bytes);
                let scale = col.type_info.scale.unwrap_or(0) as u32;

                #[cfg(feature = "decimal")]
                {
                    use rust_decimal::Decimal;
                    // rust_decimal holds 96-bit mantissas with scale <= 28;
                    // SQL Server NUMERIC goes to 38 digits, so legitimate
                    // wire values can exceed it. That must be an error, not
                    // a silent fall back to f64 (~15-16 significant digits):
                    // a lossy value read, written back, or compared
                    // downstream corrupts data (issue #157).
                    let decimal = i128::try_from(mantissa)
                        .ok()
                        .and_then(|m| Decimal::try_from_i128_with_scale(m, scale).ok());
                    match decimal {
                        Some(mut decimal) => {
                            if sign == 0 {
                                decimal.set_sign_negative(true);
                            }
                            SqlValue::Decimal(decimal)
                        }
                        None => {
                            return Err(mssql_types::TypeError::InvalidDecimal(format!(
                                "NUMERIC value (mantissa {mantissa}, scale {scale}) exceeds \
                                 rust_decimal's 96-bit/scale-28 range; CAST the column to a \
                                 narrower NUMERIC, FLOAT, or VARCHAR in the query"
                            ))
                            .into());
                        }
                    }
                }

                #[cfg(not(feature = "decimal"))]
                {
                    // Without the decimal feature, convert to f64
                    let divisor = 10f64.powi(scale as i32);
                    let value = (mantissa as f64) / divisor;
                    let value = if sign == 0 { -value } else { value };
                    SqlValue::Double(value)
                }
            }
        }

        // DATETIME/SMALLDATETIME nullable (1-byte length prefix)
        TypeId::DateTimeN => {
            if buf.remaining() < 1 {
                return Err(Error::Protocol(
                    "unexpected EOF reading DateTimeN length".into(),
                ));
            }
            let len = buf.get_u8() as usize;
            if len == 0 {
                SqlValue::Null
            } else if buf.remaining() < len {
                return Err(Error::Protocol("unexpected EOF reading DateTimeN".into()));
            } else {
                match len {
                    4 => {
                        // SMALLDATETIME: 2 bytes days + 2 bytes minutes
                        let days = buf.get_u16_le() as i64;
                        let minutes = buf.get_u16_le() as u32;
                        #[cfg(feature = "chrono")]
                        {
                            SqlValue::DateTime(smalldatetime_from_wire(days, minutes)?)
                        }
                        #[cfg(not(feature = "chrono"))]
                        {
                            SqlValue::String(format!("SMALLDATETIME({days},{minutes})"))
                        }
                    }
                    8 => {
                        // DATETIME: 4 bytes days + 4 bytes 1/300ths of second
                        let days = buf.get_i32_le() as i64;
                        let time_300ths = buf.get_u32_le() as u64;
                        #[cfg(feature = "chrono")]
                        {
                            SqlValue::DateTime(datetime_from_wire(days, time_300ths)?)
                        }
                        #[cfg(not(feature = "chrono"))]
                        {
                            SqlValue::String(format!("DATETIME({days},{time_300ths})"))
                        }
                    }
                    _ => {
                        return Err(Error::Protocol(format!("invalid DateTimeN length: {len}")));
                    }
                }
            }
        }

        // Fixed DATETIME (8 bytes)
        TypeId::DateTime => {
            if buf.remaining() < 8 {
                return Err(Error::Protocol("unexpected EOF reading DATETIME".into()));
            }
            let days = buf.get_i32_le() as i64;
            let time_300ths = buf.get_u32_le() as u64;
            #[cfg(feature = "chrono")]
            {
                SqlValue::DateTime(datetime_from_wire(days, time_300ths)?)
            }
            #[cfg(not(feature = "chrono"))]
            {
                SqlValue::String(format!("DATETIME({days},{time_300ths})"))
            }
        }

        // Fixed SMALLDATETIME (4 bytes)
        TypeId::DateTime4 => {
            if buf.remaining() < 4 {
                return Err(Error::Protocol(
                    "unexpected EOF reading SMALLDATETIME".into(),
                ));
            }
            let days = buf.get_u16_le() as i64;
            let minutes = buf.get_u16_le() as u32;
            #[cfg(feature = "chrono")]
            {
                SqlValue::DateTime(smalldatetime_from_wire(days, minutes)?)
            }
            #[cfg(not(feature = "chrono"))]
            {
                SqlValue::String(format!("SMALLDATETIME({days},{minutes})"))
            }
        }

        // DATE (3 bytes, nullable with 1-byte length prefix)
        TypeId::Date => {
            if buf.remaining() < 1 {
                return Err(Error::Protocol("unexpected EOF reading DATE length".into()));
            }
            let len = buf.get_u8() as usize;
            if len == 0 {
                SqlValue::Null
            } else if len != 3 {
                return Err(Error::Protocol(format!("invalid DATE length: {len}")));
            } else if buf.remaining() < 3 {
                return Err(Error::Protocol("unexpected EOF reading DATE".into()));
            } else {
                // 3 bytes little-endian days since 0001-01-01
                let days = buf.get_u8() as u32
                    | ((buf.get_u8() as u32) << 8)
                    | ((buf.get_u8() as u32) << 16);
                #[cfg(feature = "chrono")]
                {
                    let base = chrono::NaiveDate::from_ymd_opt(1, 1, 1)
                        .expect("epoch 0001-01-01 is valid");
                    let date = base + chrono::Duration::days(days as i64);
                    SqlValue::Date(date)
                }
                #[cfg(not(feature = "chrono"))]
                {
                    SqlValue::String(format!("DATE({days})"))
                }
            }
        }

        // TIME (variable length with scale, 1-byte length prefix)
        TypeId::Time => {
            if buf.remaining() < 1 {
                return Err(Error::Protocol("unexpected EOF reading TIME length".into()));
            }
            let len = buf.get_u8() as usize;
            if len == 0 {
                SqlValue::Null
            } else if buf.remaining() < len {
                return Err(Error::Protocol("unexpected EOF reading TIME".into()));
            } else {
                let mut time_bytes = [0u8; 8];
                for byte in time_bytes.iter_mut().take(len) {
                    *byte = buf.get_u8();
                }
                let intervals = u64::from_le_bytes(time_bytes);
                #[cfg(feature = "chrono")]
                {
                    let scale = col.type_info.scale.unwrap_or(7);
                    let time = intervals_to_time(intervals, scale);
                    SqlValue::Time(time)
                }
                #[cfg(not(feature = "chrono"))]
                {
                    SqlValue::String(format!("TIME({intervals})"))
                }
            }
        }

        // DATETIME2 (variable length: TIME bytes + 3 bytes date, 1-byte length prefix)
        TypeId::DateTime2 => {
            if buf.remaining() < 1 {
                return Err(Error::Protocol(
                    "unexpected EOF reading DATETIME2 length".into(),
                ));
            }
            let len = buf.get_u8() as usize;
            if len == 0 {
                SqlValue::Null
            } else if buf.remaining() < len {
                return Err(Error::Protocol("unexpected EOF reading DATETIME2".into()));
            } else {
                let scale = col.type_info.scale.unwrap_or(7);
                let time_len = time_bytes_for_scale(scale);
                // Reads below are driven by scale metadata, not by `len`:
                // a short declared length must be an error, not a panic.
                if len < time_len + 3 {
                    return Err(Error::Protocol(format!(
                        "DATETIME2 length {len} too short for scale {scale}"
                    )));
                }

                // Read time
                let mut time_bytes = [0u8; 8];
                for byte in time_bytes.iter_mut().take(time_len) {
                    *byte = buf.get_u8();
                }
                let intervals = u64::from_le_bytes(time_bytes);

                // Read date (3 bytes)
                let days = buf.get_u8() as u32
                    | ((buf.get_u8() as u32) << 8)
                    | ((buf.get_u8() as u32) << 16);

                #[cfg(feature = "chrono")]
                {
                    let base = chrono::NaiveDate::from_ymd_opt(1, 1, 1)
                        .expect("epoch 0001-01-01 is valid");
                    let date = base + chrono::Duration::days(days as i64);
                    let time = intervals_to_time(intervals, scale);
                    SqlValue::DateTime(date.and_time(time))
                }
                #[cfg(not(feature = "chrono"))]
                {
                    SqlValue::String(format!("DATETIME2({days},{intervals})"))
                }
            }
        }

        // DATETIMEOFFSET (variable length: TIME bytes + 3 bytes date + 2 bytes offset)
        TypeId::DateTimeOffset => {
            if buf.remaining() < 1 {
                return Err(Error::Protocol(
                    "unexpected EOF reading DATETIMEOFFSET length".into(),
                ));
            }
            let len = buf.get_u8() as usize;
            if len == 0 {
                SqlValue::Null
            } else if buf.remaining() < len {
                return Err(Error::Protocol(
                    "unexpected EOF reading DATETIMEOFFSET".into(),
                ));
            } else {
                let scale = col.type_info.scale.unwrap_or(7);
                let time_len = time_bytes_for_scale(scale);
                // Reads below are driven by scale metadata, not by `len`:
                // a short declared length must be an error, not a panic.
                if len < time_len + 5 {
                    return Err(Error::Protocol(format!(
                        "DATETIMEOFFSET length {len} too short for scale {scale}"
                    )));
                }

                // Read time
                let mut time_bytes = [0u8; 8];
                for byte in time_bytes.iter_mut().take(time_len) {
                    *byte = buf.get_u8();
                }
                let intervals = u64::from_le_bytes(time_bytes);

                // Read date (3 bytes)
                let days = buf.get_u8() as u32
                    | ((buf.get_u8() as u32) << 8)
                    | ((buf.get_u8() as u32) << 16);

                // Read offset in minutes (2 bytes, signed)
                let offset_minutes = buf.get_i16_le();

                #[cfg(feature = "chrono")]
                {
                    use chrono::TimeZone;
                    let base = chrono::NaiveDate::from_ymd_opt(1, 1, 1)
                        .expect("epoch 0001-01-01 is valid");
                    let date = base + chrono::Duration::days(days as i64);
                    let time = intervals_to_time(intervals, scale);
                    let offset = chrono::FixedOffset::east_opt((offset_minutes as i32) * 60)
                        .unwrap_or_else(|| {
                            chrono::FixedOffset::east_opt(0).expect("UTC offset 0 is valid")
                        });
                    // The wire date/time portion is UTC per MS-TDS §2.2.5.5.1.9;
                    // attach the offset without shifting the instant.
                    let datetime = offset.from_utc_datetime(&date.and_time(time));
                    SqlValue::DateTimeOffset(datetime)
                }
                #[cfg(not(feature = "chrono"))]
                {
                    SqlValue::String(format!(
                        "DATETIMEOFFSET({days},{intervals},{offset_minutes})"
                    ))
                }
            }
        }

        // TEXT type - always uses PLP encoding (deprecated LOB type)
        TypeId::Text => parse_plp_varchar(buf, col.type_info.collation.as_ref())?,

        // Legacy byte-length string types (Char, VarChar) - 1-byte length prefix
        TypeId::Char | TypeId::VarChar => {
            if buf.remaining() < 1 {
                return Err(Error::Protocol(
                    "unexpected EOF reading legacy varchar length".into(),
                ));
            }
            let len = buf.get_u8();
            if len == 0xFF {
                SqlValue::Null
            } else if len == 0 {
                SqlValue::String(String::new())
            } else if buf.remaining() < len as usize {
                return Err(Error::Protocol(
                    "unexpected EOF reading legacy varchar data".into(),
                ));
            } else {
                let data = &buf[..len as usize];
                // Use collation-aware decoding for non-ASCII text
                let s = decode_varchar_string(data, col.type_info.collation.as_ref());
                buf.advance(len as usize);
                SqlValue::String(s)
            }
        }

        // Variable-length string types (BigVarChar, BigChar)
        TypeId::BigVarChar | TypeId::BigChar => {
            // Check if this is a MAX type (uses PLP encoding)
            if col.type_info.max_length == Some(0xFFFF) {
                // PLP format: 8-byte total length, then chunks
                parse_plp_varchar(buf, col.type_info.collation.as_ref())?
            } else {
                // 2-byte length prefix for non-MAX types
                if buf.remaining() < 2 {
                    return Err(Error::Protocol(
                        "unexpected EOF reading varchar length".into(),
                    ));
                }
                let len = buf.get_u16_le();
                if len == 0xFFFF {
                    SqlValue::Null
                } else if buf.remaining() < len as usize {
                    return Err(Error::Protocol(
                        "unexpected EOF reading varchar data".into(),
                    ));
                } else {
                    let data = &buf[..len as usize];
                    // Use collation-aware decoding for non-ASCII text
                    let s = decode_varchar_string(data, col.type_info.collation.as_ref());
                    buf.advance(len as usize);
                    SqlValue::String(s)
                }
            }
        }

        // NTEXT type - always uses PLP encoding (deprecated LOB type)
        TypeId::NText => parse_plp_nvarchar(buf)?,

        // Variable-length Unicode string types (NVarChar, NChar)
        TypeId::NVarChar | TypeId::NChar => {
            // Check if this is a MAX type (uses PLP encoding)
            if col.type_info.max_length == Some(0xFFFF) {
                // PLP format: 8-byte total length, then chunks
                parse_plp_nvarchar(buf)?
            } else {
                // 2-byte length prefix (in bytes, not chars) for non-MAX types
                if buf.remaining() < 2 {
                    return Err(Error::Protocol(
                        "unexpected EOF reading nvarchar length".into(),
                    ));
                }
                let len = buf.get_u16_le();
                if len == 0xFFFF {
                    SqlValue::Null
                } else if buf.remaining() < len as usize {
                    return Err(Error::Protocol(
                        "unexpected EOF reading nvarchar data".into(),
                    ));
                } else {
                    let data = &buf[..len as usize];
                    // UTF-16LE to String
                    let utf16: Vec<u16> = data
                        .chunks_exact(2)
                        .map(|chunk| u16::from_le_bytes([chunk[0], chunk[1]]))
                        .collect();
                    let s = String::from_utf16(&utf16)
                        .map_err(|_| Error::Protocol("invalid UTF-16 in nvarchar".into()))?;
                    buf.advance(len as usize);
                    SqlValue::String(s)
                }
            }
        }

        // IMAGE type - always uses PLP encoding (deprecated LOB type)
        TypeId::Image => parse_plp_varbinary(buf)?,

        // Legacy byte-length binary types (Binary, VarBinary) - 1-byte length prefix
        TypeId::Binary | TypeId::VarBinary => {
            if buf.remaining() < 1 {
                return Err(Error::Protocol(
                    "unexpected EOF reading legacy varbinary length".into(),
                ));
            }
            let len = buf.get_u8();
            if len == 0xFF {
                SqlValue::Null
            } else if len == 0 {
                SqlValue::Binary(bytes::Bytes::new())
            } else if buf.remaining() < len as usize {
                return Err(Error::Protocol(
                    "unexpected EOF reading legacy varbinary data".into(),
                ));
            } else {
                let data = bytes::Bytes::copy_from_slice(&buf[..len as usize]);
                buf.advance(len as usize);
                SqlValue::Binary(data)
            }
        }

        // Variable-length binary types (BigVarBinary, BigBinary)
        TypeId::BigVarBinary | TypeId::BigBinary => {
            // Check if this is a MAX type (uses PLP encoding)
            if col.type_info.max_length == Some(0xFFFF) {
                // PLP format: 8-byte total length, then chunks
                parse_plp_varbinary(buf)?
            } else {
                if buf.remaining() < 2 {
                    return Err(Error::Protocol(
                        "unexpected EOF reading varbinary length".into(),
                    ));
                }
                let len = buf.get_u16_le();
                if len == 0xFFFF {
                    SqlValue::Null
                } else if buf.remaining() < len as usize {
                    return Err(Error::Protocol(
                        "unexpected EOF reading varbinary data".into(),
                    ));
                } else {
                    let data = bytes::Bytes::copy_from_slice(&buf[..len as usize]);
                    buf.advance(len as usize);
                    SqlValue::Binary(data)
                }
            }
        }

        // XML type - always uses PLP encoding
        TypeId::Xml => {
            // Parse as PLP NVARCHAR (XML is UTF-16 encoded in TDS)
            match parse_plp_nvarchar(buf)? {
                SqlValue::Null => SqlValue::Null,
                SqlValue::String(s) => SqlValue::Xml(s),
                _ => {
                    return Err(Error::Protocol(
                        "unexpected value type when parsing XML".into(),
                    ));
                }
            }
        }

        // GUID/UniqueIdentifier
        TypeId::Guid => {
            if buf.remaining() < 1 {
                return Err(Error::Protocol("unexpected EOF reading GUID length".into()));
            }
            let len = buf.get_u8();
            if len == 0 {
                SqlValue::Null
            } else if len != 16 {
                return Err(Error::Protocol(format!("invalid GUID length: {len}")));
            } else if buf.remaining() < 16 {
                return Err(Error::Protocol("unexpected EOF reading GUID".into()));
            } else {
                // SQL Server stores GUIDs in mixed-endian format:
                // first 3 groups byte-swapped, last 2 groups big-endian.
                // Swap back to RFC 4122 big-endian format.
                decode_guid_bytes(buf)
            }
        }

        // SQL_VARIANT - contains embedded type info
        TypeId::Variant => parse_sql_variant(buf)?,

        // UDT (User-Defined Type) - uses PLP encoding, return as binary
        TypeId::Udt => parse_plp_varbinary(buf)?,

        // Default: treat as binary with 2-byte length prefix
        _ => {
            // Try to read as variable-length with 2-byte length
            if buf.remaining() < 2 {
                return Err(Error::Protocol(format!(
                    "unexpected EOF reading {:?}",
                    col.type_id
                )));
            }
            let len = buf.get_u16_le();
            if len == 0xFFFF {
                SqlValue::Null
            } else if buf.remaining() < len as usize {
                return Err(Error::Protocol(format!(
                    "unexpected EOF reading {:?} data",
                    col.type_id
                )));
            } else {
                let data = bytes::Bytes::copy_from_slice(&buf[..len as usize]);
                buf.advance(len as usize);
                SqlValue::Binary(data)
            }
        }
    };

    Ok(value)
}

/// Parse PLP-encoded NVARCHAR(MAX) data.
///
/// PLP format stored by decode_plp_type:
/// - 8-byte total length (0xFFFFFFFFFFFFFFFF = NULL)
/// - Chunks: 4-byte chunk length + chunk data, terminated by 0 length
pub(crate) fn parse_plp_nvarchar(buf: &mut &[u8]) -> Result<SqlValue> {
    if buf.remaining() < 8 {
        return Err(Error::Protocol(
            "unexpected EOF reading PLP total length".into(),
        ));
    }

    let total_len = buf.get_u64_le();
    if total_len == 0xFFFFFFFFFFFFFFFF {
        return Ok(SqlValue::Null);
    }

    // Read all chunks and concatenate the data
    let mut all_data = Vec::new();
    loop {
        if buf.remaining() < 4 {
            return Err(Error::Protocol(
                "unexpected EOF reading PLP chunk length".into(),
            ));
        }
        let chunk_len = buf.get_u32_le() as usize;
        if chunk_len == 0 {
            break; // End of PLP data
        }
        if buf.remaining() < chunk_len {
            return Err(Error::Protocol(
                "unexpected EOF reading PLP chunk data".into(),
            ));
        }
        all_data.extend_from_slice(&buf[..chunk_len]);
        buf.advance(chunk_len);
    }

    // Convert UTF-16LE to String
    let utf16: Vec<u16> = all_data
        .chunks_exact(2)
        .map(|chunk| u16::from_le_bytes([chunk[0], chunk[1]]))
        .collect();
    let s = String::from_utf16(&utf16)
        .map_err(|_| Error::Protocol("invalid UTF-16 in PLP nvarchar".into()))?;
    Ok(SqlValue::String(s))
}

/// Decode VARCHAR bytes to a String using collation-aware encoding.
///
/// When the `encoding` feature is enabled and a collation is provided,
/// this decodes the bytes using the appropriate character encoding based
/// on the collation's LCID. Otherwise falls back to UTF-8 lossy conversion.
#[allow(unused_variables)]
fn decode_varchar_string(data: &[u8], collation: Option<&Collation>) -> String {
    // Try collation-aware decoding first (handles GBK, Shift-JIS, etc.)
    #[cfg(feature = "encoding")]
    if let Some(coll) = collation {
        if let Some(encoding) = coll.encoding() {
            let (decoded, _, had_errors) = encoding.decode(data);
            if !had_errors {
                return decoded.into_owned();
            }
        }
    }

    // Fallback: lossy UTF-8 conversion
    String::from_utf8_lossy(data).into_owned()
}

/// Parse PLP-encoded VARCHAR(MAX) data.
fn parse_plp_varchar(buf: &mut &[u8], collation: Option<&Collation>) -> Result<SqlValue> {
    if buf.remaining() < 8 {
        return Err(Error::Protocol(
            "unexpected EOF reading PLP total length".into(),
        ));
    }

    let total_len = buf.get_u64_le();
    if total_len == 0xFFFFFFFFFFFFFFFF {
        return Ok(SqlValue::Null);
    }

    // Read all chunks and concatenate the data
    let mut all_data = Vec::new();
    loop {
        if buf.remaining() < 4 {
            return Err(Error::Protocol(
                "unexpected EOF reading PLP chunk length".into(),
            ));
        }
        let chunk_len = buf.get_u32_le() as usize;
        if chunk_len == 0 {
            break; // End of PLP data
        }
        if buf.remaining() < chunk_len {
            return Err(Error::Protocol(
                "unexpected EOF reading PLP chunk data".into(),
            ));
        }
        all_data.extend_from_slice(&buf[..chunk_len]);
        buf.advance(chunk_len);
    }

    // Decode using collation-aware encoding
    let s = decode_varchar_string(&all_data, collation);
    Ok(SqlValue::String(s))
}

/// Parse PLP-encoded VARBINARY(MAX) data.
pub(crate) fn parse_plp_varbinary(buf: &mut &[u8]) -> Result<SqlValue> {
    if buf.remaining() < 8 {
        return Err(Error::Protocol(
            "unexpected EOF reading PLP total length".into(),
        ));
    }

    let total_len = buf.get_u64_le();
    if total_len == 0xFFFFFFFFFFFFFFFF {
        return Ok(SqlValue::Null);
    }

    // Read all chunks and concatenate the data
    let mut all_data = Vec::new();
    loop {
        if buf.remaining() < 4 {
            return Err(Error::Protocol(
                "unexpected EOF reading PLP chunk length".into(),
            ));
        }
        let chunk_len = buf.get_u32_le() as usize;
        if chunk_len == 0 {
            break; // End of PLP data
        }
        if buf.remaining() < chunk_len {
            return Err(Error::Protocol(
                "unexpected EOF reading PLP chunk data".into(),
            ));
        }
        all_data.extend_from_slice(&buf[..chunk_len]);
        buf.advance(chunk_len);
    }

    Ok(SqlValue::Binary(bytes::Bytes::from(all_data)))
}

/// Parse SQL_VARIANT data which contains embedded type information.
///
/// SQL_VARIANT format:
/// - 4 bytes: total length (0 = NULL)
/// - 1 byte: base type ID
/// - 1 byte: property byte count
/// - N bytes: type-specific properties
/// - Remaining bytes: actual data
fn parse_sql_variant(buf: &mut &[u8]) -> Result<SqlValue> {
    // Read 4-byte length
    if buf.remaining() < 4 {
        return Err(Error::Protocol(
            "unexpected EOF reading SQL_VARIANT length".into(),
        ));
    }
    let total_len = buf.get_u32_le() as usize;

    if total_len == 0 {
        return Ok(SqlValue::Null);
    }

    if buf.remaining() < total_len {
        return Err(Error::Protocol(
            "unexpected EOF reading SQL_VARIANT data".into(),
        ));
    }

    // Read type info
    if total_len < 2 {
        return Err(Error::Protocol(
            "SQL_VARIANT too short for type info".into(),
        ));
    }

    let base_type = buf.get_u8();
    let prop_count = buf.get_u8() as usize;

    if buf.remaining() < prop_count {
        return Err(Error::Protocol(
            "unexpected EOF reading SQL_VARIANT properties".into(),
        ));
    }

    // Data length is total_len - 2 (type, prop_count) - prop_count
    let data_len = total_len.saturating_sub(2).saturating_sub(prop_count);

    // Parse based on base type
    // See MS-TDS SQL_VARIANT specification for type mappings
    match base_type {
        // Integer types (no properties)
        0x30 => {
            // TINYINT
            buf.advance(prop_count);
            if data_len < 1 {
                return Ok(SqlValue::Null);
            }
            let v = buf.get_u8();
            Ok(SqlValue::TinyInt(v))
        }
        0x32 => {
            // BIT
            buf.advance(prop_count);
            if data_len < 1 {
                return Ok(SqlValue::Null);
            }
            let v = buf.get_u8();
            Ok(SqlValue::Bool(v != 0))
        }
        0x34 => {
            // SMALLINT
            buf.advance(prop_count);
            if data_len < 2 {
                return Ok(SqlValue::Null);
            }
            let v = buf.get_i16_le();
            Ok(SqlValue::SmallInt(v))
        }
        0x38 => {
            // INT
            buf.advance(prop_count);
            if data_len < 4 {
                return Ok(SqlValue::Null);
            }
            let v = buf.get_i32_le();
            Ok(SqlValue::Int(v))
        }
        0x7F => {
            // BIGINT
            buf.advance(prop_count);
            if data_len < 8 {
                return Ok(SqlValue::Null);
            }
            let v = buf.get_i64_le();
            Ok(SqlValue::BigInt(v))
        }
        0x6D => {
            // FLOATN - 1 prop byte (length)
            let float_len = if prop_count >= 1 { buf.get_u8() } else { 8 };
            buf.advance(prop_count.saturating_sub(1));

            if float_len == 4 && data_len >= 4 {
                let v = buf.get_f32_le();
                Ok(SqlValue::Float(v))
            } else if data_len >= 8 {
                let v = buf.get_f64_le();
                Ok(SqlValue::Double(v))
            } else {
                Ok(SqlValue::Null)
            }
        }
        0x6E => {
            // MONEYN - 1 prop byte (length)
            let money_len = if prop_count >= 1 { buf.get_u8() } else { 8 };
            buf.advance(prop_count.saturating_sub(1));

            if money_len == 0 || data_len == 0 {
                Ok(SqlValue::Null)
            } else if (money_len == 4 && data_len >= 4) || (money_len == 8 && data_len >= 8) {
                parse_money_value(buf, money_len as usize)
            } else {
                buf.advance(data_len);
                Ok(SqlValue::Null)
            }
        }
        0x6F => {
            // DATETIMEN - 1 prop byte (length)
            #[cfg(feature = "chrono")]
            let dt_len = if prop_count >= 1 { buf.get_u8() } else { 8 };
            #[cfg(not(feature = "chrono"))]
            if prop_count >= 1 {
                buf.get_u8();
            }
            buf.advance(prop_count.saturating_sub(1));

            #[cfg(feature = "chrono")]
            {
                if dt_len == 4 && data_len >= 4 {
                    // SMALLDATETIME
                    let days = buf.get_u16_le() as i64;
                    let mins = buf.get_u16_le() as u32;
                    Ok(SqlValue::DateTime(smalldatetime_from_wire(days, mins)?))
                } else if data_len >= 8 {
                    // DATETIME
                    let days = buf.get_i32_le() as i64;
                    let ticks = buf.get_u32_le() as u64;
                    Ok(SqlValue::DateTime(datetime_from_wire(days, ticks)?))
                } else {
                    Ok(SqlValue::Null)
                }
            }
            #[cfg(not(feature = "chrono"))]
            {
                buf.advance(data_len);
                Ok(SqlValue::Null)
            }
        }
        0x6A | 0x6C => {
            // DECIMALN/NUMERICN - 2 prop bytes (precision, scale)
            let _precision = if prop_count >= 1 { buf.get_u8() } else { 18 };
            let scale = if prop_count >= 2 { buf.get_u8() } else { 0 };
            buf.advance(prop_count.saturating_sub(2));

            if data_len < 1 {
                return Ok(SqlValue::Null);
            }

            let sign = buf.get_u8();
            let mantissa_len = data_len - 1;

            if mantissa_len > 16 {
                // Too large, skip and return null
                buf.advance(mantissa_len);
                return Ok(SqlValue::Null);
            }

            let mut mantissa_bytes = [0u8; 16];
            for i in 0..mantissa_len.min(16) {
                mantissa_bytes[i] = buf.get_u8();
            }
            let mantissa = u128::from_le_bytes(mantissa_bytes);

            #[cfg(feature = "decimal")]
            {
                use rust_decimal::Decimal;
                // Same overflow class as the top-level DECIMAL branch:
                // rust_decimal holds 96-bit mantissas with scale <= 28, but a
                // 16-byte wire mantissa (legitimate 38-digit NUMERIC or
                // hostile) can exceed that regardless of scale. The old
                // `scale > 28` guard did not cover an oversized mantissa, so
                // `from_i128_with_scale` could panic. Fall back to f64 on any
                // out-of-range value.
                let decimal = i128::try_from(mantissa)
                    .ok()
                    .and_then(|m| Decimal::try_from_i128_with_scale(m, scale as u32).ok());
                match decimal {
                    Some(mut decimal) => {
                        if sign == 0 {
                            decimal.set_sign_negative(true);
                        }
                        Ok(SqlValue::Decimal(decimal))
                    }
                    None => Err(mssql_types::TypeError::InvalidDecimal(format!(
                        "NUMERIC value in sql_variant (mantissa {mantissa}, scale {scale}) \
                         exceeds rust_decimal's 96-bit/scale-28 range; CAST the column to a \
                         narrower NUMERIC, FLOAT, or VARCHAR in the query"
                    ))
                    .into()),
                }
            }
            #[cfg(not(feature = "decimal"))]
            {
                let divisor = 10f64.powi(scale as i32);
                let value = (mantissa as f64) / divisor;
                let value = if sign == 0 { -value } else { value };
                Ok(SqlValue::Double(value))
            }
        }
        0x24 => {
            // UNIQUEIDENTIFIER (no properties)
            buf.advance(prop_count);
            if data_len < 16 {
                return Ok(SqlValue::Null);
            }
            // SQL Server stores GUIDs in mixed-endian format — swap back to RFC 4122
            Ok(decode_guid_bytes(buf))
        }
        0x28 => {
            // DATE (no properties)
            buf.advance(prop_count);
            #[cfg(feature = "chrono")]
            {
                if data_len < 3 {
                    return Ok(SqlValue::Null);
                }
                let mut date_bytes = [0u8; 4];
                date_bytes[0] = buf.get_u8();
                date_bytes[1] = buf.get_u8();
                date_bytes[2] = buf.get_u8();
                let days = u32::from_le_bytes(date_bytes);
                let base =
                    chrono::NaiveDate::from_ymd_opt(1, 1, 1).expect("epoch 0001-01-01 is valid");
                let date = base + chrono::Duration::days(days as i64);
                Ok(SqlValue::Date(date))
            }
            #[cfg(not(feature = "chrono"))]
            {
                buf.advance(data_len);
                Ok(SqlValue::Null)
            }
        }
        0x29 => {
            // TIME - 1 prop byte (scale)
            #[cfg_attr(not(feature = "chrono"), allow(unused_variables))]
            let scale = if prop_count >= 1 { buf.get_u8() } else { 7 };
            buf.advance(prop_count.saturating_sub(1));

            #[cfg(feature = "chrono")]
            {
                if data_len == 0 {
                    return Ok(SqlValue::Null);
                }
                let time_len = time_bytes_for_scale(scale);
                if data_len < time_len {
                    return Ok(SqlValue::Null);
                }
                let mut time_bytes = [0u8; 8];
                for byte in time_bytes.iter_mut().take(time_len) {
                    *byte = buf.get_u8();
                }
                // Consume any remaining data bytes beyond the time portion
                if data_len > time_len {
                    buf.advance(data_len - time_len);
                }
                let intervals = u64::from_le_bytes(time_bytes);
                Ok(SqlValue::Time(intervals_to_time(intervals, scale)))
            }
            #[cfg(not(feature = "chrono"))]
            {
                buf.advance(data_len);
                Ok(SqlValue::Null)
            }
        }
        0x2A => {
            // DATETIME2 - 1 prop byte (scale)
            #[cfg_attr(not(feature = "chrono"), allow(unused_variables))]
            let scale = if prop_count >= 1 { buf.get_u8() } else { 7 };
            buf.advance(prop_count.saturating_sub(1));

            #[cfg(feature = "chrono")]
            {
                let time_len = time_bytes_for_scale(scale);
                if data_len < time_len + 3 {
                    return Ok(SqlValue::Null);
                }

                let mut time_bytes = [0u8; 8];
                for byte in time_bytes.iter_mut().take(time_len) {
                    *byte = buf.get_u8();
                }
                let intervals = u64::from_le_bytes(time_bytes);

                let days = buf.get_u8() as u32
                    | ((buf.get_u8() as u32) << 8)
                    | ((buf.get_u8() as u32) << 16);

                // Consume any remaining data bytes
                let consumed = time_len + 3;
                if data_len > consumed {
                    buf.advance(data_len - consumed);
                }

                let base =
                    chrono::NaiveDate::from_ymd_opt(1, 1, 1).expect("epoch 0001-01-01 is valid");
                let date = base + chrono::Duration::days(days as i64);
                let time = intervals_to_time(intervals, scale);
                Ok(SqlValue::DateTime(date.and_time(time)))
            }
            #[cfg(not(feature = "chrono"))]
            {
                buf.advance(data_len);
                Ok(SqlValue::Null)
            }
        }
        0x2B => {
            // DATETIMEOFFSET - 1 prop byte (scale)
            #[cfg_attr(not(feature = "chrono"), allow(unused_variables))]
            let scale = if prop_count >= 1 { buf.get_u8() } else { 7 };
            buf.advance(prop_count.saturating_sub(1));

            #[cfg(feature = "chrono")]
            {
                let time_len = time_bytes_for_scale(scale);
                if data_len < time_len + 3 + 2 {
                    return Ok(SqlValue::Null);
                }

                let mut time_bytes = [0u8; 8];
                for byte in time_bytes.iter_mut().take(time_len) {
                    *byte = buf.get_u8();
                }
                let intervals = u64::from_le_bytes(time_bytes);

                let days = buf.get_u8() as u32
                    | ((buf.get_u8() as u32) << 8)
                    | ((buf.get_u8() as u32) << 16);

                let offset_minutes = buf.get_i16_le();

                // Consume any remaining data bytes
                let consumed = time_len + 3 + 2;
                if data_len > consumed {
                    buf.advance(data_len - consumed);
                }

                use chrono::TimeZone;
                let base =
                    chrono::NaiveDate::from_ymd_opt(1, 1, 1).expect("epoch 0001-01-01 is valid");
                let date = base + chrono::Duration::days(days as i64);
                let time = intervals_to_time(intervals, scale);
                let offset = chrono::FixedOffset::east_opt((offset_minutes as i32) * 60)
                    .unwrap_or_else(|| {
                        chrono::FixedOffset::east_opt(0).expect("UTC offset 0 is valid")
                    });
                // The wire date/time portion is UTC per MS-TDS §2.2.5.5.1.9;
                // attach the offset without shifting the instant.
                let datetime = offset.from_utc_datetime(&date.and_time(time));
                Ok(SqlValue::DateTimeOffset(datetime))
            }
            #[cfg(not(feature = "chrono"))]
            {
                buf.advance(data_len);
                Ok(SqlValue::Null)
            }
        }
        0xA7 | 0x2F | 0x27 => {
            // BigVarChar/BigChar/VarChar/Char - 7 prop bytes (collation 5 + maxlen 2)
            // Parse collation from property bytes (5 bytes: 4 LCID + 1 sort_id)
            let collation = if prop_count >= 5 && buf.remaining() >= 5 {
                let lcid = buf.get_u32_le();
                let sort_id = buf.get_u8();
                buf.advance(prop_count.saturating_sub(5)); // Skip remaining props (max_length)
                Some(Collation { lcid, sort_id })
            } else {
                buf.advance(prop_count);
                None
            };
            if data_len == 0 {
                return Ok(SqlValue::String(String::new()));
            }
            let data = &buf[..data_len];
            // Use collation-aware decoding for non-ASCII text
            let s = decode_varchar_string(data, collation.as_ref());
            buf.advance(data_len);
            Ok(SqlValue::String(s))
        }
        0xE7 | 0xEF => {
            // NVarChar/NChar - 7 prop bytes (collation 5 + maxlen 2)
            buf.advance(prop_count);
            if data_len == 0 {
                return Ok(SqlValue::String(String::new()));
            }
            // UTF-16LE encoded
            let utf16: Vec<u16> = buf[..data_len]
                .chunks_exact(2)
                .map(|chunk| u16::from_le_bytes([chunk[0], chunk[1]]))
                .collect();
            buf.advance(data_len);
            let s = String::from_utf16(&utf16)
                .map_err(|_| Error::Protocol("invalid UTF-16 in SQL_VARIANT nvarchar".into()))?;
            Ok(SqlValue::String(s))
        }
        0xA5 | 0x2D | 0x25 => {
            // BigVarBinary/BigBinary/Binary/VarBinary - 2 prop bytes (maxlen)
            buf.advance(prop_count);
            let data = bytes::Bytes::copy_from_slice(&buf[..data_len]);
            buf.advance(data_len);
            Ok(SqlValue::Binary(data))
        }
        _ => {
            // Unknown type - return as binary
            buf.advance(prop_count);
            let data = bytes::Bytes::copy_from_slice(&buf[..data_len]);
            buf.advance(data_len);
            Ok(SqlValue::Binary(data))
        }
    }
}

/// Calculate number of bytes needed for TIME based on scale.
fn time_bytes_for_scale(scale: u8) -> usize {
    match scale {
        0..=2 => 3,
        3..=4 => 4,
        5..=7 => 5,
        _ => 5, // Default to max precision
    }
}

/// Convert 100-nanosecond intervals to NaiveTime.
#[cfg(feature = "chrono")]
fn intervals_to_time(intervals: u64, scale: u8) -> chrono::NaiveTime {
    // Scale determines the unit:
    // scale 0: seconds
    // scale 1: 100ms
    // scale 2: 10ms
    // scale 3: 1ms
    // scale 4: 100us
    // scale 5: 10us
    // scale 6: 1us
    // scale 7: 100ns
    // Saturating: `intervals` comes from the wire, and a hostile value must
    // not overflow-panic in debug builds (saturation lands in the
    // out-of-range fallback below).
    let nanos = match scale {
        0 => intervals.saturating_mul(1_000_000_000),
        1 => intervals.saturating_mul(100_000_000),
        2 => intervals.saturating_mul(10_000_000),
        3 => intervals.saturating_mul(1_000_000),
        4 => intervals.saturating_mul(100_000),
        5 => intervals.saturating_mul(10_000),
        6 => intervals.saturating_mul(1_000),
        7 => intervals.saturating_mul(100),
        _ => intervals.saturating_mul(100),
    };

    let secs = (nanos / 1_000_000_000) as u32;
    let nano_part = (nanos % 1_000_000_000) as u32;

    chrono::NaiveTime::from_num_seconds_from_midnight_opt(secs, nano_part)
        .unwrap_or_else(|| chrono::NaiveTime::from_hms_opt(0, 0, 0).expect("midnight is valid"))
}

/// Decode 16 GUID bytes from SQL Server mixed-endian wire format to RFC 4122 format.
///
/// SQL Server stores UUIDs with the first 3 groups byte-swapped (little-endian)
/// and the last 2 groups in big-endian order. This function reads 16 bytes,
/// swaps the first 3 groups back, and returns the appropriate SqlValue.
fn decode_guid_bytes(buf: &mut &[u8]) -> SqlValue {
    let mut bytes = [0u8; 16];

    // First 4 bytes — little-endian on wire, swap to big-endian
    bytes[3] = buf.get_u8();
    bytes[2] = buf.get_u8();
    bytes[1] = buf.get_u8();
    bytes[0] = buf.get_u8();

    // Next 2 bytes — little-endian on wire, swap to big-endian
    bytes[5] = buf.get_u8();
    bytes[4] = buf.get_u8();

    // Next 2 bytes — little-endian on wire, swap to big-endian
    bytes[7] = buf.get_u8();
    bytes[6] = buf.get_u8();

    // Last 8 bytes — big-endian, keep as-is
    for byte in &mut bytes[8..16] {
        *byte = buf.get_u8();
    }

    #[cfg(feature = "uuid")]
    {
        SqlValue::Uuid(uuid::Uuid::from_bytes(bytes))
    }
    #[cfg(not(feature = "uuid"))]
    {
        SqlValue::Binary(bytes::Bytes::copy_from_slice(&bytes))
    }
}

#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::panic)]
mod tests {
    use super::*;
    use tds_protocol::token::TypeInfo;

    // ========================================================================
    // PLP (Partially Length-Prefixed) Parsing Tests
    // ========================================================================
    //
    // These tests verify that MAX type (NVARCHAR(MAX), VARCHAR(MAX), VARBINARY(MAX))
    // data is correctly parsed from the PLP wire format.

    /// Helper to create PLP data with a single chunk.
    fn make_plp_data(total_len: u64, chunks: &[&[u8]]) -> Vec<u8> {
        let mut data = Vec::new();
        // 8-byte total length
        data.extend_from_slice(&total_len.to_le_bytes());
        // Chunks
        for chunk in chunks {
            let len = chunk.len() as u32;
            data.extend_from_slice(&len.to_le_bytes());
            data.extend_from_slice(chunk);
        }
        // Terminating zero-length chunk
        data.extend_from_slice(&0u32.to_le_bytes());
        data
    }

    #[test]
    fn test_parse_plp_nvarchar_simple() {
        // "Hello" in UTF-16LE: H=0x0048, e=0x0065, l=0x006C, l=0x006C, o=0x006F
        let utf16_data = [0x48, 0x00, 0x65, 0x00, 0x6C, 0x00, 0x6C, 0x00, 0x6F, 0x00];
        let plp = make_plp_data(10, &[&utf16_data]);
        let mut buf: &[u8] = &plp;

        let result = parse_plp_nvarchar(&mut buf).unwrap();
        match result {
            SqlValue::String(s) => assert_eq!(s, "Hello"),
            _ => panic!("expected String, got {result:?}"),
        }
    }

    #[test]
    fn test_parse_plp_nvarchar_null() {
        // NULL is indicated by total_len = 0xFFFFFFFFFFFFFFFF
        let plp = 0xFFFFFFFFFFFFFFFFu64.to_le_bytes();
        let mut buf: &[u8] = &plp;

        let result = parse_plp_nvarchar(&mut buf).unwrap();
        assert!(matches!(result, SqlValue::Null));
    }

    #[test]
    fn test_parse_plp_nvarchar_empty() {
        // Empty string: total_len=0, single zero-length chunk
        let plp = make_plp_data(0, &[]);
        let mut buf: &[u8] = &plp;

        let result = parse_plp_nvarchar(&mut buf).unwrap();
        match result {
            SqlValue::String(s) => assert_eq!(s, ""),
            _ => panic!("expected empty String"),
        }
    }

    #[test]
    fn test_parse_plp_nvarchar_multi_chunk() {
        // "Hello" split across two chunks: "Hel" + "lo"
        let chunk1 = [0x48, 0x00, 0x65, 0x00, 0x6C, 0x00]; // "Hel"
        let chunk2 = [0x6C, 0x00, 0x6F, 0x00]; // "lo"
        let plp = make_plp_data(10, &[&chunk1, &chunk2]);
        let mut buf: &[u8] = &plp;

        let result = parse_plp_nvarchar(&mut buf).unwrap();
        match result {
            SqlValue::String(s) => assert_eq!(s, "Hello"),
            _ => panic!("expected String"),
        }
    }

    #[test]
    fn test_parse_plp_varchar_simple() {
        let data = b"Hello World";
        let plp = make_plp_data(11, &[data]);
        let mut buf: &[u8] = &plp;

        let result = parse_plp_varchar(&mut buf, None).unwrap();
        match result {
            SqlValue::String(s) => assert_eq!(s, "Hello World"),
            _ => panic!("expected String"),
        }
    }

    #[test]
    fn test_parse_plp_varchar_null() {
        let plp = 0xFFFFFFFFFFFFFFFFu64.to_le_bytes();
        let mut buf: &[u8] = &plp;

        let result = parse_plp_varchar(&mut buf, None).unwrap();
        assert!(matches!(result, SqlValue::Null));
    }

    #[test]
    fn test_parse_plp_varbinary_simple() {
        let data = [0x01, 0x02, 0x03, 0x04, 0x05];
        let plp = make_plp_data(5, &[&data]);
        let mut buf: &[u8] = &plp;

        let result = parse_plp_varbinary(&mut buf).unwrap();
        match result {
            SqlValue::Binary(b) => assert_eq!(&b[..], &[0x01, 0x02, 0x03, 0x04, 0x05]),
            _ => panic!("expected Binary"),
        }
    }

    #[test]
    fn test_parse_plp_varbinary_null() {
        let plp = 0xFFFFFFFFFFFFFFFFu64.to_le_bytes();
        let mut buf: &[u8] = &plp;

        let result = parse_plp_varbinary(&mut buf).unwrap();
        assert!(matches!(result, SqlValue::Null));
    }

    #[test]
    fn test_parse_plp_varbinary_large() {
        // Test with larger data split across multiple chunks
        let chunk1: Vec<u8> = (0..100u8).collect();
        let chunk2: Vec<u8> = (100..200u8).collect();
        let chunk3: Vec<u8> = (200..255u8).collect();
        let total_len = chunk1.len() + chunk2.len() + chunk3.len();
        let plp = make_plp_data(total_len as u64, &[&chunk1, &chunk2, &chunk3]);
        let mut buf: &[u8] = &plp;

        let result = parse_plp_varbinary(&mut buf).unwrap();
        match result {
            SqlValue::Binary(b) => {
                assert_eq!(b.len(), 255);
                // Verify data integrity
                for (i, &byte) in b.iter().enumerate() {
                    assert_eq!(byte, i as u8);
                }
            }
            _ => panic!("expected Binary"),
        }
    }

    // ========================================================================
    // Multi-Column Row Parsing Tests
    // ========================================================================
    //
    // These tests verify that parsing multiple columns in a row works correctly,
    // especially for scenarios where string columns are followed by integer columns.

    /// Build raw row data for a non-MAX NVarChar followed by an IntN.
    /// This mimics the scenario: SELECT @name AS greeting, @value AS number
    fn make_nvarchar_int_row(nvarchar_value: &str, int_value: i32) -> Vec<u8> {
        let mut data = Vec::new();

        // Column 0: NVarChar (non-MAX) - 2-byte length prefix (in bytes)
        let utf16: Vec<u16> = nvarchar_value.encode_utf16().collect();
        let byte_len = (utf16.len() * 2) as u16;
        data.extend_from_slice(&byte_len.to_le_bytes());
        for code_unit in utf16 {
            data.extend_from_slice(&code_unit.to_le_bytes());
        }

        // Column 1: IntN - 1-byte length prefix
        data.push(4); // 4 bytes for INT
        data.extend_from_slice(&int_value.to_le_bytes());

        data
    }

    // ========================================================================
    // Hostile wire data: out-of-range date/time values from a malicious or
    // buggy server must produce protocol errors, never panics.
    // ========================================================================

    #[cfg(feature = "chrono")]
    fn datetime_col(type_id: TypeId, col_type: u8, max_length: Option<u32>) -> ColumnData {
        ColumnData {
            name: "c".to_string(),
            type_id,
            col_type,
            flags: 0x01,
            user_type: 0,
            type_info: TypeInfo {
                max_length,
                precision: None,
                scale: None,
                collation: None,
            },
            crypto_metadata: None,
        }
    }

    #[cfg(feature = "chrono")]
    #[test]
    fn hostile_smalldatetime_minutes_is_error_not_panic() {
        // DateTimeN, len=4: days=0, minutes=0xFFFF (>= 1440 is invalid)
        let data = [4u8, 0x00, 0x00, 0xFF, 0xFF];
        let col = datetime_col(TypeId::DateTimeN, 0x6F, Some(4));
        let mut buf: &[u8] = &data;
        assert!(parse_column_value(&mut buf, &col).is_err());

        // Fixed SMALLDATETIME (DateTime4): same payload, no length prefix
        let data = [0x00, 0x00, 0xFF, 0xFF];
        let col = datetime_col(TypeId::DateTime4, 0x3A, None);
        let mut buf: &[u8] = &data;
        assert!(parse_column_value(&mut buf, &col).is_err());
    }

    #[cfg(feature = "chrono")]
    #[test]
    fn hostile_datetime_days_overflow_is_error_not_panic() {
        // DateTimeN, len=8: days=i32::MAX overflows chrono's date range
        let mut data = vec![8u8];
        data.extend_from_slice(&i32::MAX.to_le_bytes());
        data.extend_from_slice(&0u32.to_le_bytes());
        let col = datetime_col(TypeId::DateTimeN, 0x6F, Some(8));
        let mut buf: &[u8] = &data;
        assert!(parse_column_value(&mut buf, &col).is_err());

        // Fixed DATETIME: days=i32::MIN
        let mut data = Vec::new();
        data.extend_from_slice(&i32::MIN.to_le_bytes());
        data.extend_from_slice(&0u32.to_le_bytes());
        let col = datetime_col(TypeId::DateTime, 0x3D, None);
        let mut buf: &[u8] = &data;
        assert!(parse_column_value(&mut buf, &col).is_err());
    }

    #[cfg(feature = "chrono")]
    #[test]
    fn hostile_datetime_time_300ths_is_error_not_panic() {
        // DateTimeN, len=8: valid days, time_300ths=u32::MAX (> 24h of 300ths)
        let mut data = vec![8u8];
        data.extend_from_slice(&0i32.to_le_bytes());
        data.extend_from_slice(&u32::MAX.to_le_bytes());
        let col = datetime_col(TypeId::DateTimeN, 0x6F, Some(8));
        let mut buf: &[u8] = &data;
        assert!(parse_column_value(&mut buf, &col).is_err());
    }

    #[cfg(feature = "chrono")]
    #[test]
    fn hostile_truncated_n_types_are_error_not_panic() {
        // Declared length with no payload bytes behind it (found by the
        // parse_column_value fuzz target on FloatN).
        for (type_id, col_type, len) in [
            (TypeId::IntN, 0x26u8, 8u8),
            (TypeId::FloatN, 0x6D, 4),
            (TypeId::BitN, 0x68, 1),
        ] {
            let data = [len];
            let col = datetime_col(type_id, col_type, Some(len as u32));
            let mut buf: &[u8] = &data;
            assert!(
                parse_column_value(&mut buf, &col).is_err(),
                "{type_id:?} must error on truncated payload"
            );
        }
    }

    #[cfg(feature = "chrono")]
    #[test]
    fn hostile_short_datetime2_len_is_error_not_panic() {
        // DATETIME2 declared len=1 (1 byte present) but scale 7 implies
        // time_len=5 + 3 date bytes; reads are scale-driven, so a short
        // declared length must error.
        let data = [1u8, 0xAA];
        let mut col = datetime_col(TypeId::DateTime2, 0x2A, None);
        col.type_info.scale = Some(7);
        let mut buf: &[u8] = &data;
        assert!(parse_column_value(&mut buf, &col).is_err());

        // Same for DATETIMEOFFSET (time_len + 5)
        let data = [1u8, 0xAA];
        let mut col = datetime_col(TypeId::DateTimeOffset, 0x2B, None);
        col.type_info.scale = Some(7);
        let mut buf: &[u8] = &data;
        assert!(parse_column_value(&mut buf, &col).is_err());
    }

    #[cfg(feature = "chrono")]
    #[test]
    fn hostile_variant_datetime_days_overflow_is_error_not_panic() {
        // SQL_VARIANT: total_len=11, base_type=0x6F (DATETIMEN), prop_count=1,
        // dt_len=8, then days=i32::MAX + ticks=0
        let mut data = Vec::new();
        data.extend_from_slice(&11u32.to_le_bytes());
        data.push(0x6F);
        data.push(0x01);
        data.push(0x08);
        data.extend_from_slice(&i32::MAX.to_le_bytes());
        data.extend_from_slice(&0u32.to_le_bytes());
        let mut buf: &[u8] = &data;
        assert!(parse_sql_variant(&mut buf).is_err());
    }

    #[cfg(feature = "decimal")]
    #[test]
    fn hostile_variant_decimal_over_96bit_is_error_not_panic() {
        // SQL_VARIANT DECIMALN (0x6A) with a 16-byte mantissa above 2^96.
        // rust_decimal cannot hold it; this must be a descriptive error,
        // never a panic — and never a silent f64 fallback, which corrupted
        // legitimate 38-digit NUMERIC values down to ~15-16 significant
        // digits (issue #157).
        // total_len=21: base_type + prop_count + precision + scale + sign(1)
        // + mantissa(16) = 2 + 2 + 17.
        let mantissa = 1u128 << 100;
        let mut data = Vec::new();
        data.extend_from_slice(&21u32.to_le_bytes());
        data.push(0x6A); // DECIMALN
        data.push(0x02); // prop_count: precision, scale
        data.push(38); // precision
        data.push(10); // scale
        data.push(0x01); // sign (positive)
        data.extend_from_slice(&mantissa.to_le_bytes());
        let mut buf: &[u8] = &data;
        let err = parse_sql_variant(&mut buf).expect_err("oversized NUMERIC must error");
        assert!(
            err.to_string().contains("rust_decimal"),
            "error should explain the range limitation: {err}"
        );
    }

    #[cfg(feature = "chrono")]
    #[test]
    fn hostile_time_intervals_do_not_panic() {
        // intervals_to_time multiplies wire-supplied u64 by up to 1e9; must
        // not overflow-panic in debug builds. Falls back to midnight today.
        let t = intervals_to_time(u64::MAX, 0);
        let _ = t; // any non-panicking result is acceptable
    }

    #[test]
    fn test_parse_row_nvarchar_then_int() {
        // Build raw row data for: "World", 42
        let raw_data = make_nvarchar_int_row("World", 42);

        // Create column metadata
        let col0 = ColumnData {
            name: "greeting".to_string(),
            type_id: TypeId::NVarChar,
            col_type: 0xE7,
            flags: 0x01,
            user_type: 0,
            type_info: TypeInfo {
                max_length: Some(10), // 5 chars * 2 bytes = 10
                precision: None,
                scale: None,
                collation: None,
            },
            crypto_metadata: None,
        };

        let col1 = ColumnData {
            name: "number".to_string(),
            type_id: TypeId::IntN,
            col_type: 0x26,
            flags: 0x01,
            user_type: 0,
            type_info: TypeInfo {
                max_length: Some(4),
                precision: None,
                scale: None,
                collation: None,
            },
            crypto_metadata: None,
        };

        let mut buf: &[u8] = &raw_data;

        // Parse column 0 (NVarChar)
        let value0 = parse_column_value(&mut buf, &col0).unwrap();
        match value0 {
            SqlValue::String(s) => assert_eq!(s, "World"),
            _ => panic!("expected String, got {value0:?}"),
        }

        // Parse column 1 (IntN)
        let value1 = parse_column_value(&mut buf, &col1).unwrap();
        match value1 {
            SqlValue::Int(i) => assert_eq!(i, 42),
            _ => panic!("expected Int, got {value1:?}"),
        }

        // Buffer should be fully consumed
        assert_eq!(buf.len(), 0, "buffer should be fully consumed");
    }

    #[test]
    fn test_parse_row_multiple_types() {
        // Build raw data for: NULL (NVarChar), 123 (IntN), "Test" (NVarChar), NULL (IntN)
        let mut data = Vec::new();

        // Column 0: NVarChar NULL (0xFFFF)
        data.extend_from_slice(&0xFFFFu16.to_le_bytes());

        // Column 1: IntN with value 123
        data.push(4); // 4 bytes
        data.extend_from_slice(&123i32.to_le_bytes());

        // Column 2: NVarChar "Test"
        let utf16: Vec<u16> = "Test".encode_utf16().collect();
        data.extend_from_slice(&((utf16.len() * 2) as u16).to_le_bytes());
        for code_unit in utf16 {
            data.extend_from_slice(&code_unit.to_le_bytes());
        }

        // Column 3: IntN NULL (0 length)
        data.push(0);

        // Metadata for 4 columns
        let col0 = ColumnData {
            name: "col0".to_string(),
            type_id: TypeId::NVarChar,
            col_type: 0xE7,
            flags: 0x01,
            user_type: 0,
            type_info: TypeInfo {
                max_length: Some(100),
                precision: None,
                scale: None,
                collation: None,
            },
            crypto_metadata: None,
        };
        let col1 = ColumnData {
            name: "col1".to_string(),
            type_id: TypeId::IntN,
            col_type: 0x26,
            flags: 0x01,
            user_type: 0,
            type_info: TypeInfo {
                max_length: Some(4),
                precision: None,
                scale: None,
                collation: None,
            },
            crypto_metadata: None,
        };
        let col2 = col0.clone();
        let col3 = col1.clone();

        let mut buf: &[u8] = &data;

        // Parse all 4 columns
        let v0 = parse_column_value(&mut buf, &col0).unwrap();
        assert!(matches!(v0, SqlValue::Null), "col0 should be Null");

        let v1 = parse_column_value(&mut buf, &col1).unwrap();
        assert!(matches!(v1, SqlValue::Int(123)), "col1 should be 123");

        let v2 = parse_column_value(&mut buf, &col2).unwrap();
        match v2 {
            SqlValue::String(s) => assert_eq!(s, "Test"),
            _ => panic!("col2 should be 'Test'"),
        }

        let v3 = parse_column_value(&mut buf, &col3).unwrap();
        assert!(matches!(v3, SqlValue::Null), "col3 should be Null");

        // Buffer should be fully consumed
        assert_eq!(buf.len(), 0, "buffer should be fully consumed");
    }

    #[test]
    fn test_parse_row_with_unicode() {
        // Test with Unicode characters that need proper UTF-16 encoding
        let test_str = "Héllo Wörld 日本語";
        let mut data = Vec::new();

        // NVarChar with Unicode
        let utf16: Vec<u16> = test_str.encode_utf16().collect();
        data.extend_from_slice(&((utf16.len() * 2) as u16).to_le_bytes());
        for code_unit in utf16 {
            data.extend_from_slice(&code_unit.to_le_bytes());
        }

        // IntN value
        data.push(8); // BIGINT
        data.extend_from_slice(&9999999999i64.to_le_bytes());

        let col0 = ColumnData {
            name: "text".to_string(),
            type_id: TypeId::NVarChar,
            col_type: 0xE7,
            flags: 0x01,
            user_type: 0,
            type_info: TypeInfo {
                max_length: Some(100),
                precision: None,
                scale: None,
                collation: None,
            },
            crypto_metadata: None,
        };
        let col1 = ColumnData {
            name: "num".to_string(),
            type_id: TypeId::IntN,
            col_type: 0x26,
            flags: 0x01,
            user_type: 0,
            type_info: TypeInfo {
                max_length: Some(8),
                precision: None,
                scale: None,
                collation: None,
            },
            crypto_metadata: None,
        };

        let mut buf: &[u8] = &data;

        let v0 = parse_column_value(&mut buf, &col0).unwrap();
        match v0 {
            SqlValue::String(s) => assert_eq!(s, test_str),
            _ => panic!("expected String"),
        }

        let v1 = parse_column_value(&mut buf, &col1).unwrap();
        match v1 {
            SqlValue::BigInt(i) => assert_eq!(i, 9999999999),
            _ => panic!("expected BigInt"),
        }

        assert_eq!(buf.len(), 0, "buffer should be fully consumed");
    }
}