tds_driver/tds/codec/

column_data.rs

mod binary;
mod bit;
mod bytes_mut_with_type_info;
#[cfg(feature = "tds73")]
mod date;
#[cfg(feature = "tds73")]
mod datetime2;
mod datetimen;
#[cfg(feature = "tds73")]
mod datetimeoffsetn;
mod fixed_len;
mod float;
mod guid;
mod image;
mod int;
mod money;
mod plp;
mod sql_variant;
mod string;
mod text;
#[cfg(feature = "tds73")]
mod time;
mod udt;
mod var_len;
mod xml;

/// Upper bound on how many bytes a value decoder will *pre-allocate* from a
/// server-supplied length field before it has read the corresponding data.
///
/// The wire length is untrusted: a malformed or hostile server can claim a
/// value is up to `u32::MAX`/`u64::MAX` bytes long. Reserving that up front is a
/// memory-exhaustion vector (and, for `u64` lengths, can even exceed `Vec`'s
/// `isize::MAX` capacity limit and panic). Decoders therefore cap the initial
/// reservation to this value and let the buffer grow as bytes actually arrive;
/// a short/lying length still fails cleanly when the read runs out of input.
pub(crate) const MAX_PREALLOC: usize = 8192; // 8 KiB

/// Absolute ceiling on the *total* size of a single PLP (partially
/// length-prefixed) value — `varchar(max)`, `nvarchar(max)`, `varbinary(max)`,
/// `xml`, and CLR UDTs. SQL Server's own MAX types top out at `2^31 - 1` bytes,
/// so any value that would grow past this is malformed. Without this bound the
/// "unknown length" PLP form (which streams an arbitrary number of chunks until
/// a zero-length terminator) lets a hostile server grow the accumulation buffer
/// without limit and OOM the client on a single column value.
pub(crate) const MAX_PLP_SIZE: usize = i32::MAX as usize;

use super::{Encode, FixedLenType, TypeInfo, VarLenContext, VarLenType};
use crate::tds::collation::Collation;
#[cfg(feature = "tds73")]
use crate::tds::time::{Date, DateTime2, DateTimeOffset, Time};
use crate::{
    tds::{time::DateTime, time::SmallDateTime, xml::XmlData, Numeric},
    FromSql, FromSqlOwned, IntoSql, SqlReadBytes, ToSql,
};
use bytes::{BufMut, BytesMut};
pub(crate) use bytes_mut_with_type_info::BytesMutWithTypeInfo;
use std::borrow::{BorrowMut, Cow};
use uuid::Uuid;

/// UTF-16 code units in the 2^31 - 1 bytes of nvarchar(max), xml and ntext.
const NVARCHAR_MAX_UNITS: usize = i32::MAX as usize / 2;

/// Precision of every numeric parameter: the server's maximum.
const NUMERIC_PARAM_PRECISION: u8 = 38;

/// Whether `s` fits `nvarchar(4000)`, whose limit is in UTF-16 code units.
/// A UTF-8 byte is at most one unit and a unit at most three bytes, so only
/// strings of 4001..=12000 bytes need counting.
fn fits_nvarchar_4000(s: &str) -> bool {
    s.len() <= 4000 || (s.len() <= 12000 && s.encode_utf16().count() <= 4000)
}

const PLP_NULL: u64 = 0xffffffffffffffff;

/// `s`'s UTF-16 length if over `max` units. A unit is at least one UTF-8
/// byte, so only longer strings are counted.
fn utf16_units_over(s: &str, max: usize) -> Option<usize> {
    if s.len() <= max {
        return None;
    }
    Some(s.encode_utf16().count()).filter(|&units| units > max)
}

/// `len` as the length of a (max), text, ntext or image value, which holds
/// at most 2^31 - 1 bytes.
fn lob_len(len: usize) -> Option<u32> {
    i32::try_from(len).ok().map(i32::unsigned_abs)
}

/// An nvarchar(max) or xml value: UTF-16 in one PLP chunk, of a total
/// length not known before encoding.
pub(crate) fn put_plp_utf16(dst: &mut BytesMut, s: &str) -> crate::Result<()> {
    let too_long = |units: usize| {
        crate::Error::Conversion(
            format!(
                "a string of {units} UTF-16 code units is longer than nvarchar(max) and xml \
                 hold ({NVARCHAR_MAX_UNITS})"
            )
            .into(),
        )
    };
    // Before writing what can be gigabytes.
    if let Some(units) = utf16_units_over(s, NVARCHAR_MAX_UNITS) {
        return Err(too_long(units));
    }

    dst.put_u64_le(0xfffffffffffffffe); // unknown length
    let len_pos = dst.len();
    dst.put_u32_le(0);
    for unit in s.encode_utf16() {
        dst.put_u16_le(unit);
    }

    let bytes = dst.len() - len_pos - 4;
    let len = lob_len(bytes).ok_or_else(|| too_long(bytes / 2))?;
    dst[len_pos..len_pos + 4].copy_from_slice(&len.to_le_bytes());

    // A zero-length chunk is the terminator: an empty value is that alone.
    if len > 0 {
        dst.put_u32_le(0);
    }

    Ok(())
}

/// `nvarchar(max)` TYPE_INFO and value.
/// A `varchar` parameter: TYPE_INFO in the session's `collation`, then the
/// value in that collation's code page. `varchar(8000)`, or `varchar(max)` as
/// PLP when `max`; a NULL is `varchar(8000)` either way, unless `max` asks for
/// an OUT parameter's full width.
fn put_varchar(
    dst: &mut BytesMutWithTypeInfo<'_>,
    s: Option<&str>,
    collation: Option<Collation>,
    max: bool,
) -> crate::Result<()> {
    let collation = collation.ok_or_else(|| {
        crate::Error::Encoding(
            "a varchar parameter needs the session collation, which the server has not sent".into(),
        )
    })?;
    let bytes = s
        .map(|s| string::encode_varchar(s, &collation))
        .transpose()?;

    dst.put_u8(VarLenType::BigVarChar as u8);
    dst.put_u16_le(if max { 0xffff } else { 8000 });
    dst.put_u32_le(collation.info());
    dst.put_u8(collation.sort_id());

    match (bytes, max) {
        (Some(bytes), false) => {
            // `type_name` promises this: UTF-8 is never shorter than the code page.
            let len = u16::try_from(bytes.len()).map_err(|_| {
                crate::Error::Encoding(
                    format!("{} bytes do not fit varchar(8000)", bytes.len()).into(),
                )
            })?;
            dst.put_u16_le(len);
            dst.extend_from_slice(&bytes);
        }
        (Some(bytes), true) => put_plp(dst, &bytes),
        (None, false) => dst.put_u16_le(0xffff),
        (None, true) => dst.put_u64_le(PLP_NULL),
    }

    Ok(())
}

fn put_nvarchar_max(dst: &mut BytesMut, s: Option<&str>) -> crate::Result<()> {
    // length: 0xffff and raw collation
    dst.put_u8(VarLenType::NVarchar as u8);
    dst.extend_from_slice(&[0xff_u8; 2]);
    dst.extend_from_slice(&[0u8; 5]);

    match s {
        Some(s) => put_plp_utf16(dst, s),
        None => {
            dst.put_u64_le(PLP_NULL);
            Ok(())
        }
    }
}

/// `varbinary(max)` TYPE_INFO and value.
fn put_varbinary_max(dst: &mut BytesMut, bytes: Option<&[u8]>) -> crate::Result<()> {
    dst.put_u8(VarLenType::BigVarBin as u8);
    // Max length
    dst.put_u16_le(0xffff_u16);

    let Some(bytes) = bytes else {
        dst.put_u64_le(PLP_NULL);
        return Ok(());
    };
    let len = lob_len(bytes.len()).ok_or_else(|| {
        crate::Error::Conversion(
            format!(
                "{} bytes are more than varbinary(max) holds ({})",
                bytes.len(),
                i32::MAX
            )
            .into(),
        )
    })?;

    // Unknown length, then the value in one chunk.
    dst.put_u64_le(0xfffffffffffffffe_u64);
    dst.put_u32_le(len);

    if len > 0 {
        dst.extend_from_slice(bytes);
        // PLP_TERMINATOR
        dst.put_u32_le(0);
    }

    Ok(())
}

/// Number of days between `0001-01-01` (the `DateTime2`/`Date` epoch) and
/// `1900-01-01` (the `datetime`/`Datetimen` epoch).
#[cfg(feature = "tds73")]
const DAYS_YEAR_1_TO_1900: u32 = 693_595;

/// Time of day unit for converting between `datetime` (1/300 s),
/// `smalldatetime` (minutes) and `datetime2` (10^-7 s) without rounding twice.
const UNITS_PER_SECOND: u64 = 3_000_000_000;
const UNITS_PER_DAY: u64 = 86_400 * UNITS_PER_SECOND;

fn not_null_error() -> crate::Error {
    crate::Error::BulkInput("NULL for a NOT NULL column".into())
}

/// Bulk/TVP integers and floats go out at the column's width: the server
/// reads the column's type from the bytes, so a narrower value is misread.
fn encode_number(
    data: &ColumnData<'_>,
    ti: &TypeInfo,
    dst: &mut BytesMutWithTypeInfo<'_>,
) -> Option<crate::Result<()>> {
    let (float, width, nullable) = match ti {
        TypeInfo::FixedLen(FixedLenType::Int1) => (false, 1, false),
        TypeInfo::FixedLen(FixedLenType::Int2) => (false, 2, false),
        TypeInfo::FixedLen(FixedLenType::Int4) => (false, 4, false),
        TypeInfo::FixedLen(FixedLenType::Int8) => (false, 8, false),
        TypeInfo::FixedLen(FixedLenType::Float4) => (true, 4, false),
        TypeInfo::FixedLen(FixedLenType::Float8) => (true, 8, false),
        TypeInfo::VarLenSized(vlc) if vlc.r#type() == VarLenType::Intn => (false, vlc.len(), true),
        TypeInfo::VarLenSized(vlc) if vlc.r#type() == VarLenType::Floatn => (true, vlc.len(), true),
        _ => return None,
    };

    let bytes = if float {
        let value = match *data {
            ColumnData::F32(v) => v.map(f64::from),
            ColumnData::F64(v) => v,
            _ => return None,
        };
        value.map(|v| float_bytes(v, width))
    } else {
        let value = match *data {
            ColumnData::U8(v) => v.map(i64::from),
            ColumnData::I16(v) => v.map(i64::from),
            ColumnData::I32(v) => v.map(i64::from),
            ColumnData::I64(v) => v,
            _ => return None,
        };
        value.map(|v| int_bytes(v, width))
    };

    Some(match bytes.transpose() {
        Err(e) => Err(e),
        Ok(None) if nullable => {
            dst.put_u8(0);
            Ok(())
        }
        Ok(None) => Err(not_null_error()),
        Ok(Some(bytes)) => {
            if nullable {
                dst.put_u8(width as u8);
            }
            dst.extend_from_slice(&bytes[..width]);
            Ok(())
        }
    })
}

fn int_bytes(value: i64, width: usize) -> crate::Result<[u8; 8]> {
    let (name, fits) = match width {
        1 => ("tinyint", (0..=255).contains(&value)),
        2 => ("smallint", i16::try_from(value).is_ok()),
        4 => ("int", i32::try_from(value).is_ok()),
        8 => ("bigint", true),
        _ => {
            return Err(crate::Error::Protocol(
                format!("integer column of {width} bytes").into(),
            ))
        }
    };

    if !fits {
        return Err(crate::Error::Conversion(
            format!("{value} is out of range for {name}").into(),
        ));
    }

    Ok(value.to_le_bytes())
}

/// The server rejects a NaN or infinite float with its generic error 8023,
/// "the RPC stream is incorrect"; this says what is wrong before sending.
fn finite(value: f64) -> crate::Result<f64> {
    if value.is_finite() {
        Ok(value)
    } else {
        Err(crate::Error::Conversion(
            format!("float: SQL Server has no NaN or infinity, got {value}").into(),
        ))
    }
}

fn float_bytes(value: f64, width: usize) -> crate::Result<[u8; 8]> {
    let value = finite(value)?;
    let mut bytes = [0u8; 8];

    match width {
        8 => bytes = value.to_le_bytes(),
        4 => {
            let narrowed = value as f32;
            if value.is_finite() && narrowed.is_infinite() {
                return Err(crate::Error::Conversion(
                    format!("{value:e} is out of range for real").into(),
                ));
            }
            bytes[..4].copy_from_slice(&narrowed.to_le_bytes());
        }
        _ => {
            return Err(crate::Error::Protocol(
                format!("float column of {width} bytes").into(),
            ))
        }
    }

    Ok(bytes)
}

/// Bulk/TVP `datetime`/`smalldatetime` values go out in the column's own
/// format, rounded the way SQL Server's CAST rounds them.
fn encode_legacy_datetime(
    data: &ColumnData<'_>,
    ti: &TypeInfo,
    dst: &mut BytesMutWithTypeInfo<'_>,
) -> Option<crate::Result<()>> {
    let (small, nullable) = match ti {
        TypeInfo::FixedLen(FixedLenType::Datetime) => (false, false),
        TypeInfo::FixedLen(FixedLenType::Datetime4) => (true, false),
        TypeInfo::VarLenSized(vlc) if vlc.r#type() == VarLenType::Datetimen => {
            (vlc.len() == 4, true)
        }
        _ => return None,
    };

    // Days since 1900-01-01 and the time of day in `UNITS_PER_SECOND`.
    let parts = match *data {
        // Validated: rounding would carry 24:00 and later into the next day.
        ColumnData::DateTime(v) => v.map(|v| {
            v.validate()?;
            Ok((
                i64::from(v.days()),
                u64::from(v.seconds_fragments()) * (UNITS_PER_SECOND / 300),
            ))
        }),
        // `seconds_fragments` of a smalldatetime counts minutes.
        ColumnData::SmallDateTime(v) => v.map(|v| {
            v.validate()?;
            Ok((
                i64::from(v.days()),
                u64::from(v.seconds_fragments()) * 60 * UNITS_PER_SECOND,
            ))
        }),
        #[cfg(feature = "tds73")]
        ColumnData::DateTime2(v) => v.map(datetime2_parts),
        _ => return None,
    };

    Some(match parts.transpose() {
        Err(e) => Err(e),
        Ok(None) if nullable => {
            dst.put_u8(0);
            Ok(())
        }
        Ok(None) => Err(not_null_error()),
        Ok(Some((days, units))) if small => to_smalldatetime(days, units).and_then(|v| {
            if nullable {
                dst.put_u8(4);
            }
            v.encode(dst)
        }),
        Ok(Some((days, units))) => to_datetime(days, units).and_then(|v| {
            if nullable {
                dst.put_u8(8);
            }
            v.encode(dst)
        }),
    })
}

/// Rounds half up to 1/300 s, carrying into the next day.
fn to_datetime(days: i64, units: u64) -> crate::Result<DateTime> {
    const PER_FRAGMENT: u64 = UNITS_PER_SECOND / 300;
    const FRAGMENTS_PER_DAY: u64 = UNITS_PER_DAY / PER_FRAGMENT;

    let fragments = (units + PER_FRAGMENT / 2) / PER_FRAGMENT;
    let days = days + (fragments / FRAGMENTS_PER_DAY) as i64;

    // 1753-01-01 ..= 9999-12-31
    if !(-53_690..=2_958_463).contains(&days) {
        return Err(crate::Error::Conversion(
            "value is outside the datetime range of 1753-01-01 to 9999-12-31".into(),
        ));
    }

    Ok(DateTime::new(
        days as i32,
        (fragments % FRAGMENTS_PER_DAY) as u32,
    ))
}

/// Rounds half up to the minute, carrying into the next day.
fn to_smalldatetime(days: i64, units: u64) -> crate::Result<SmallDateTime> {
    const PER_MINUTE: u64 = 60 * UNITS_PER_SECOND;

    let minutes = (units + PER_MINUTE / 2) / PER_MINUTE;
    let days = days + (minutes / 1440) as i64;

    // 1900-01-01 ..= 2079-06-06
    let days = u16::try_from(days).map_err(|_| {
        crate::Error::Conversion(
            "value is outside the smalldatetime range of 1900-01-01 to 2079-06-06".into(),
        )
    })?;

    Ok(SmallDateTime::new(days, (minutes % 1440) as u16))
}

#[cfg(feature = "tds73")]
fn datetime2_parts(dt2: DateTime2) -> crate::Result<(i64, u64)> {
    let time = rescale_time(dt2.time(), 7)?;
    let days = i64::from(dt2.date().days()) - i64::from(DAYS_YEAR_1_TO_1900);

    Ok((days, time.increments() * (UNITS_PER_SECOND / 10_000_000)))
}

/// A bulk/TVP datetime2 for a datetimeoffset or date column, as SqlBulkCopy
/// takes a DateTime for both: `new DateTimeOffset(DateTime)` in
/// SqlParameter.CoerceValue (SqlParameter.cs:2372), and WriteDate keeps the
/// days (TdsParser.cs:8141), dotnet/SqlClient@5394363. The offset is +00:00,
/// as `CAST(datetime2 AS datetimeoffset)` gives and `DateTime<Utc>` means,
/// not the local offset SqlClient gives an unspecified DateTime.
#[cfg(feature = "tds73")]
fn datetime2_for_column<'a>(data: ColumnData<'a>, ti: Option<&TypeInfo>) -> ColumnData<'a> {
    match (data, ti) {
        (ColumnData::DateTime2(v), Some(TypeInfo::VarLenSized(vlc)))
            if vlc.r#type() == VarLenType::DatetimeOffsetn =>
        {
            ColumnData::DateTimeOffset(v.map(|v| DateTimeOffset::new(v, 0)))
        }
        (ColumnData::DateTime2(v), Some(TypeInfo::VarLenSized(vlc)))
            if vlc.r#type() == VarLenType::Daten =>
        {
            ColumnData::Date(v.map(DateTime2::date))
        }
        (data, _) => data,
    }
}

/// Puts `time` at the column's scale. Dropped digits are truncated, as
/// SqlClient does for bulk copy (`value.Ticks / TICKS_FROM_SCALE[scale]` in
/// TdsParser.WriteTime and WriteDateTime2, TdsParser.cs:8172 and :8200) and
/// TVPs (TdsValueSetter.cs:588, :660, :694), dotnet/SqlClient@5394363.
#[cfg(feature = "tds73")]
fn rescale_time(time: Time, scale: u8) -> crate::Result<Time> {
    time.validate()?;

    if scale > 7 {
        return Err(crate::Error::Conversion(
            format!("time scale {scale} is invalid").into(),
        ));
    }

    // Validated: under 864 * 10^9, so times 10^7 fits.
    let increments = if scale >= time.scale() {
        time.increments() * 10u64.pow(u32::from(scale - time.scale()))
    } else {
        time.increments() / 10u64.pow(u32::from(time.scale() - scale))
    };

    Ok(Time::new(increments, scale))
}

/// Puts a bulk/TVP decimal at its column's scale, rounding half away from
/// zero as SqlDecimal.AdjustScale does, then checks it against the column's
/// precision, as SqlBulkCopy.ConvertValue does (SqlBulkCopy.cs:1816-1823,
/// dotnet/SqlClient@5394363): the server fails the whole insert (9803) on a
/// value the column cannot hold.
fn rescale_numeric(num: Numeric, precision: u8, scale: u8) -> crate::Result<Numeric> {
    // Not `num`'s Display: a deserialized Numeric can have scale > 38.
    let out_of_range = || {
        crate::Error::Conversion(
            format!(
                "{} at scale {} is out of range for decimal({precision},{scale})",
                num.value(),
                num.scale()
            )
            .into(),
        )
    };

    // At most 38 digits, as for parameters.
    let magnitude = num.value().unsigned_abs();
    if magnitude >= 10u128.pow(38) || scale > 38 {
        return Err(out_of_range());
    }

    let magnitude = match scale.checked_sub(num.scale()) {
        Some(up) => magnitude.checked_mul(10u128.pow(u32::from(up))),
        None => 10u128
            .checked_pow(u32::from(num.scale() - scale))
            .map(|f| magnitude / f + u128::from(magnitude % f >= f / 2)),
    }
    .filter(|&m| m < 10u128.pow(u32::from(precision.min(38))))
    .ok_or_else(out_of_range)?;

    // Below 10^38, so within i128.
    let value = magnitude as i128;
    Ok(Numeric::new_with_scale(
        if num.value() < 0 { -value } else { value },
        scale,
    ))
}

/// Largest value a column takes, in bytes. (max) columns declare 0xffff.
fn column_limit(vlc: &VarLenContext) -> usize {
    if vlc.len() < 0xffff {
        vlc.len()
    } else {
        MAX_PLP_SIZE
    }
}

/// One PLP chunk holds the whole value: `column_limit` keeps it below 2^31.
fn put_plp(dst: &mut BytesMutWithTypeInfo<'_>, bytes: &[u8]) {
    dst.put_u64_le(0xfffffffffffffffe); // unknown length
    dst.put_u32_le(bytes.len() as u32);

    if !bytes.is_empty() {
        dst.extend_from_slice(bytes);
        dst.put_u32_le(0); // terminator
    }
}

/// Encodes in the column's code page; the server stores the bytes as is.
fn encode_in_collation(s: &str, vlc: &VarLenContext) -> crate::Result<Vec<u8>> {
    let collation = vlc
        .collation()
        .ok_or_else(|| crate::Error::BulkInput("character column without a collation".into()))?;
    string::encode_varchar(s, &collation)
}

#[derive(Clone, Debug, PartialEq)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
/// A container of a value that can be represented as a TDS value.
pub enum ColumnData<'a> {
    /// 8-bit integer, unsigned.
    U8(Option<u8>),
    /// 16-bit integer, signed.
    I16(Option<i16>),
    /// 32-bit integer, signed.
    I32(Option<i32>),
    /// 64-bit integer, signed.
    I64(Option<i64>),
    /// 32-bit floating point number.
    F32(Option<f32>),
    /// 64-bit floating point number.
    F64(Option<f64>),
    /// Boolean.
    Bit(Option<bool>),
    /// A string value. Sent as `nvarchar`; every character column reads as
    /// this.
    String(Option<Cow<'a, str>>),
    /// A string value to send as `varchar`, in the session's collation.
    ///
    /// Use it for a parameter compared with a `varchar` column under a
    /// `SQL_*` collation: an `nvarchar` parameter outranks the column there,
    /// so the server converts every row and cannot seek the index. A
    /// character the collation's code page lacks is an error, and a value
    /// over 8000 bytes goes as `varchar(max)`.
    ///
    /// Only sent, never read: a `varchar` column arrives as
    /// [`String`](Self::String) too.
    VarChar(Option<Cow<'a, str>>),
    /// A Guid (UUID) value.
    Guid(Option<Uuid>),
    /// Binary data.
    Binary(Option<Cow<'a, [u8]>>),
    /// Numeric value (a decimal).
    Numeric(Option<Numeric>),
    /// XML data.
    Xml(Option<Cow<'a, XmlData>>),
    /// DateTime value.
    DateTime(Option<DateTime>),
    /// A small DateTime value.
    SmallDateTime(Option<SmallDateTime>),
    #[cfg(feature = "tds73")]
    #[cfg_attr(docsrs, doc(cfg(feature = "tds73")))]
    /// Time value.
    Time(Option<Time>),
    #[cfg(feature = "tds73")]
    #[cfg_attr(docsrs, doc(cfg(feature = "tds73")))]
    /// Date value.
    Date(Option<Date>),
    #[cfg(feature = "tds73")]
    #[cfg_attr(docsrs, doc(cfg(feature = "tds73")))]
    /// DateTime2 value.
    DateTime2(Option<DateTime2>),
    #[cfg(feature = "tds73")]
    #[cfg_attr(docsrs, doc(cfg(feature = "tds73")))]
    /// DateTime2 value with an offset.
    DateTimeOffset(Option<DateTimeOffset>),
}

impl<'a> ColumnData<'a> {
    pub(crate) fn type_name(&self) -> Cow<'static, str> {
        match self {
            ColumnData::U8(_) => "tinyint".into(),
            ColumnData::I16(_) => "smallint".into(),
            ColumnData::I32(_) => "int".into(),
            ColumnData::I64(_) => "bigint".into(),
            ColumnData::F32(_) => "float(24)".into(),
            ColumnData::F64(_) => "float(53)".into(),
            ColumnData::Bit(_) => "bit".into(),
            ColumnData::String(None) => "nvarchar(4000)".into(),
            ColumnData::String(Some(ref s)) if fits_nvarchar_4000(s) => "nvarchar(4000)".into(),
            // The encoder rejects a string nvarchar(max) cannot hold.
            ColumnData::String(Some(_)) => "nvarchar(max)".into(),
            // By UTF-8 length, as the encoder decides: no code page takes
            // more bytes than UTF-8, so what fits here fits on the wire.
            ColumnData::VarChar(None) => "varchar(8000)".into(),
            ColumnData::VarChar(Some(ref s)) if s.len() <= 8000 => "varchar(8000)".into(),
            ColumnData::VarChar(Some(_)) => "varchar(max)".into(),
            ColumnData::Guid(_) => "uniqueidentifier".into(),
            // A NULL is declared as a short value of its type, and as the
            // TYPE_INFO `encode` sends with it, so they share a cached plan.
            ColumnData::Binary(None) => "varbinary(8000)".into(),
            ColumnData::Binary(Some(ref b)) if b.len() <= 8000 => "varbinary(8000)".into(),
            ColumnData::Binary(_) => "varbinary(max)".into(),
            // Precision from the value would give each digit count its own
            // cached plan; 38 holds every value. The scale still comes from
            // the value, as in SqlClient, so each scale gets its own plan; a
            // NULL's is 0, SqlClient's default.
            ColumnData::Numeric(n) => format!(
                "numeric({},{})",
                NUMERIC_PARAM_PRECISION,
                n.map_or(0, Numeric::scale)
            )
            .into(),
            ColumnData::Xml(_) => "xml".into(),
            ColumnData::DateTime(_) => "datetime".into(),
            ColumnData::SmallDateTime(_) => "smalldatetime".into(),
            #[cfg(feature = "tds73")]
            ColumnData::Time(_) => "time".into(),
            #[cfg(feature = "tds73")]
            ColumnData::Date(_) => "date".into(),
            #[cfg(feature = "tds73")]
            ColumnData::DateTime2(_) => "datetime2".into(),
            #[cfg(feature = "tds73")]
            ColumnData::DateTimeOffset(_) => "datetimeoffset".into(),
        }
    }

    pub(crate) async fn decode<R>(src: &mut R, ctx: &TypeInfo) -> crate::Result<ColumnData<'a>>
    where
        R: SqlReadBytes + Unpin,
    {
        let res = match ctx {
            TypeInfo::FixedLen(fixed_ty) => fixed_len::decode(src, fixed_ty).await?,
            TypeInfo::VarLenSized(cx) => var_len::decode(src, cx).await?,
            TypeInfo::VarLenSizedPrecision { ty, scale, .. } => match ty {
                VarLenType::Decimaln | VarLenType::Numericn => {
                    ColumnData::Numeric(Numeric::decode(src, *scale).await?)
                }
                // Unreachable: `TypeInfo::decode` builds this variant for
                // decimal/numeric only.
                ty => {
                    return Err(crate::Error::Protocol(
                        format!("{ty:?} with precision and scale").into(),
                    ))
                }
            },
            TypeInfo::Xml { schema, size } => xml::decode(src, *size, schema.clone()).await?,
            TypeInfo::Udt(_) => udt::decode(src).await?,
        };

        Ok(res)
    }
}

impl<'a> ColumnData<'a> {
    /// An RPC parameter's TYPE_INFO and value.
    ///
    /// The server returns an OUT parameter in the type it was sent as, so an
    /// OUT string or binary is sent as (max): its own length would truncate
    /// the result. SqlClient sends the larger of `Size` and the value's
    /// length, or (max) for `Size = -1`, and rejects a NULL variable-length
    /// OUT parameter with no `Size`; there is no `Size` here.
    ///
    /// `collation` is the session's, for a `varchar` value.
    pub(crate) fn encode_rpc_param(
        self,
        dst: &mut BytesMut,
        out: bool,
        collation: Option<Collation>,
    ) -> crate::Result<()> {
        match self {
            ColumnData::String(s) if out => put_nvarchar_max(dst, s.as_deref())?,
            ColumnData::VarChar(s) if out => put_varchar(
                &mut BytesMutWithTypeInfo::new(dst),
                s.as_deref(),
                collation,
                true,
            )?,
            ColumnData::Binary(b) if out => put_varbinary_max(dst, b.as_deref())?,
            value => {
                return value.encode(&mut BytesMutWithTypeInfo::new(dst).with_collation(collation))
            }
        }

        Ok(())
    }
}

impl<'a> Encode<BytesMutWithTypeInfo<'a>> for ColumnData<'a> {
    fn encode(self, dst: &mut BytesMutWithTypeInfo<'a>) -> crate::Result<()> {
        if let Some(ti) = dst.type_info() {
            if let Some(res) = encode_number(&self, ti, dst) {
                return res;
            }
            if let Some(res) = encode_legacy_datetime(&self, ti, dst) {
                return res;
            }
        }

        #[cfg(feature = "tds73")]
        let data = datetime2_for_column(self, dst.type_info());
        #[cfg(not(feature = "tds73"))]
        let data = self;

        // A column (bulk load, TVP) has its own character type and collation;
        // `varchar` only means something for a parameter.
        let data = match (data, dst.type_info()) {
            (ColumnData::VarChar(s), Some(_)) => ColumnData::String(s),
            (data, _) => data,
        };

        match (data, dst.type_info()) {
            (ColumnData::VarChar(opt), None) => {
                let max = opt.as_ref().is_some_and(|s| s.len() > 8000);
                put_varchar(dst, opt.as_deref(), dst.collation(), max)?;
            }
            (ColumnData::Bit(opt), Some(TypeInfo::VarLenSized(vlc)))
                if vlc.r#type() == VarLenType::Bitn =>
            {
                if let Some(val) = opt {
                    dst.put_u8(1);
                    dst.put_u8(val as u8);
                } else {
                    dst.put_u8(0);
                }
            }
            (ColumnData::Bit(Some(val)), Some(TypeInfo::FixedLen(FixedLenType::Bit))) => {
                dst.put_u8(val as u8);
            }
            (ColumnData::Bit(opt), None) => {
                // if TypeInfo was not given, encode a TypeInfo
                // the first 1 is part of TYPE_INFO
                let header = [VarLenType::Bitn as u8, 1];
                dst.extend_from_slice(&header);
                if let Some(val) = opt {
                    // the second 1 is part of TYPE_VARBYTE
                    dst.put_u8(1);
                    dst.put_u8(val as u8);
                } else {
                    dst.put_u8(0);
                }
            }
            (ColumnData::U8(opt), None) => {
                let header = [VarLenType::Intn as u8, 1];
                dst.extend_from_slice(&header);
                if let Some(val) = opt {
                    dst.put_u8(1);
                    dst.put_u8(val);
                } else {
                    dst.put_u8(0);
                }
            }
            (ColumnData::I16(opt), None) => {
                let header = [VarLenType::Intn as u8, 2];
                dst.extend_from_slice(&header);
                if let Some(val) = opt {
                    dst.put_u8(2);
                    dst.put_i16_le(val);
                } else {
                    dst.put_u8(0);
                }
            }
            (ColumnData::I32(opt), None) => {
                let header = [VarLenType::Intn as u8, 4];
                dst.extend_from_slice(&header);
                if let Some(val) = opt {
                    dst.put_u8(4);
                    dst.put_i32_le(val);
                } else {
                    dst.put_u8(0);
                }
            }
            (ColumnData::I64(opt), None) => {
                let header = [VarLenType::Intn as u8, 8];
                dst.extend_from_slice(&header);
                if let Some(val) = opt {
                    dst.put_u8(8);
                    dst.put_i64_le(val);
                } else {
                    dst.put_u8(0);
                }
            }
            (ColumnData::F32(opt), None) => {
                let header = [VarLenType::Floatn as u8, 4];
                dst.extend_from_slice(&header);
                if let Some(val) = opt {
                    finite(val.into())?;
                    dst.put_u8(4);
                    dst.put_f32_le(val);
                } else {
                    dst.put_u8(0);
                }
            }
            (ColumnData::F64(opt), None) => {
                let header = [VarLenType::Floatn as u8, 8];
                dst.extend_from_slice(&header);
                if let Some(val) = opt {
                    finite(val)?;
                    dst.put_u8(8);
                    dst.put_f64_le(val);
                } else {
                    dst.put_u8(0);
                }
            }
            (ColumnData::F64(opt), Some(TypeInfo::VarLenSized(vlc)))
                if vlc.r#type() == VarLenType::Money =>
            {
                if let Some(val) = opt {
                    money::encode(dst, vlc.len(), val)?;
                } else {
                    dst.put_u8(0);
                }
            }
            (
                ColumnData::F64(Some(val)),
                Some(TypeInfo::FixedLen(ty @ (FixedLenType::Money | FixedLenType::Money4))),
            ) => put_fixed_money(dst, *ty, |b, len| money::encode(b, len, val))?,
            (
                ColumnData::Numeric(Some(num)),
                Some(TypeInfo::FixedLen(ty @ (FixedLenType::Money | FixedLenType::Money4))),
            ) => put_fixed_money(dst, *ty, |b, len| money::encode_numeric(b, len, num))?,
            (ColumnData::Guid(opt), Some(TypeInfo::VarLenSized(vlc)))
                if vlc.r#type() == VarLenType::Guid =>
            {
                if let Some(uuid) = opt {
                    dst.put_u8(16);

                    let mut data = *uuid.as_bytes();
                    super::guid::reorder_bytes(&mut data);
                    dst.extend_from_slice(&data);
                } else {
                    dst.put_u8(0);
                }
            }
            (ColumnData::Guid(opt), None) => {
                let header = [VarLenType::Guid as u8, 16];
                dst.extend_from_slice(&header);
                if let Some(uuid) = opt {
                    dst.put_u8(16);
                    let mut data = *uuid.as_bytes();
                    super::guid::reorder_bytes(&mut data);
                    dst.extend_from_slice(&data);
                } else {
                    dst.put_u8(0);
                }
            }
            (ColumnData::String(opt), Some(TypeInfo::VarLenSized(vlc)))
                if vlc.r#type() == VarLenType::BigChar
                    || vlc.r#type() == VarLenType::BigVarChar =>
            {
                if let Some(str) = opt {
                    let bytes = encode_in_collation(&str, vlc)?;

                    if bytes.len() > column_limit(vlc) {
                        return Err(crate::Error::BulkInput(
                            format!(
                                "Encoded string length {} exceed column limit {}",
                                bytes.len(),
                                column_limit(vlc)
                            )
                            .into(),
                        ));
                    }

                    if vlc.len() < 0xffff {
                        dst.put_u16_le(bytes.len() as u16);
                        dst.extend_from_slice(bytes.as_slice());
                    } else {
                        put_plp(dst, &bytes);
                    }
                } else if vlc.len() < 0xffff {
                    dst.put_u16_le(0xffff);
                } else {
                    dst.put_u64_le(0xffffffffffffffff)
                }
            }
            (ColumnData::String(opt), Some(TypeInfo::VarLenSized(vlc)))
                if vlc.r#type() == VarLenType::NVarchar || vlc.r#type() == VarLenType::NChar =>
            {
                if let Some(str) = opt {
                    if vlc.len() < 0xffff {
                        let len_pos = dst.len();
                        dst.put_u16_le(0u16);

                        for chr in str.encode_utf16() {
                            dst.put_u16_le(chr);
                        }

                        let length = dst.len() - len_pos - 2;

                        if length > vlc.len() {
                            return Err(crate::Error::BulkInput(
                                format!(
                                    "Encoded string length {} exceed column limit {}",
                                    length,
                                    vlc.len()
                                )
                                .into(),
                            ));
                        }

                        let dst: &mut [u8] = dst.borrow_mut();
                        let mut dst = &mut dst[len_pos..];
                        dst.put_u16_le(length as u16);
                    } else {
                        // unknown size
                        dst.put_u64_le(0xfffffffffffffffe);

                        let len_pos = dst.len();
                        dst.put_u32_le(0u32);

                        for chr in str.encode_utf16() {
                            dst.put_u16_le(chr);
                        }

                        let length = dst.len() - len_pos - 4;

                        if length > MAX_PLP_SIZE {
                            return Err(crate::Error::BulkInput(
                                format!(
                                    "Encoded string length {} exceed column limit {}",
                                    length, MAX_PLP_SIZE
                                )
                                .into(),
                            ));
                        }

                        if length > 0 {
                            // no next blob
                            dst.put_u32_le(0u32);
                        }

                        let dst: &mut [u8] = dst.borrow_mut();
                        let mut dst = &mut dst[len_pos..];
                        dst.put_u32_le(length as u32);
                    }
                } else if vlc.len() < 0xffff {
                    dst.put_u16_le(0xffff);
                } else {
                    dst.put_u64_le(0xffffffffffffffff)
                }
            }
            (ColumnData::String(opt), Some(TypeInfo::VarLenSized(vlc)))
                if vlc.r#type() == VarLenType::Text || vlc.r#type() == VarLenType::NText =>
            {
                if let Some(str) = opt {
                    if vlc.r#type() == VarLenType::Text {
                        let bytes = encode_in_collation(&str, vlc)?;
                        let len = lob_len(bytes.len()).ok_or_else(|| too_long(bytes.len()))?;

                        put_text_pointer(dst);
                        dst.put_u32_le(len);
                        dst.extend_from_slice(bytes.as_slice());
                    } else {
                        // Before writing what can be gigabytes.
                        if let Some(units) = utf16_units_over(&str, NVARCHAR_MAX_UNITS) {
                            return Err(too_long(2 * units));
                        }

                        put_text_pointer(dst);
                        let len_pos = dst.len();
                        dst.put_u32_le(0u32);
                        for chr in str.encode_utf16() {
                            dst.put_u16_le(chr);
                        }

                        let bytes = dst.len() - len_pos - 4;
                        let len = lob_len(bytes).ok_or_else(|| too_long(bytes))?;
                        dst[len_pos..len_pos + 4].copy_from_slice(&len.to_le_bytes());
                    }
                } else {
                    // NULL: zero-length text pointer
                    dst.put_u8(0);
                }
            }
            (ColumnData::Binary(opt), Some(TypeInfo::VarLenSized(vlc)))
                if vlc.r#type() == VarLenType::Image =>
            {
                if let Some(bytes) = opt {
                    let len = lob_len(bytes.len()).ok_or_else(|| too_long(bytes.len()))?;
                    put_text_pointer(dst);
                    dst.put_u32_le(len);
                    dst.extend_from_slice(&bytes);
                } else {
                    dst.put_u8(0);
                }
            }
            (ColumnData::String(Some(ref s)), None) if fits_nvarchar_4000(s) => {
                dst.put_u8(VarLenType::NVarchar as u8);
                dst.put_u16_le(8000);
                dst.extend_from_slice(&[0u8; 5][..]);

                let mut length = 0u16;
                let len_pos = dst.len();

                dst.put_u16_le(length);

                for chr in s.encode_utf16() {
                    length += 1;
                    dst.put_u16_le(chr);
                }

                let dst: &mut [u8] = dst.borrow_mut();
                let bytes = (length * 2).to_le_bytes(); // u16, two bytes

                for (i, byte) in bytes.iter().enumerate() {
                    dst[len_pos + i] = *byte;
                }
            }
            (ColumnData::String(Some(s)), None) => put_nvarchar_max(dst, Some(&s))?,
            (ColumnData::Binary(opt), Some(TypeInfo::VarLenSized(vlc)))
                if vlc.r#type() == VarLenType::BigBinary
                    || vlc.r#type() == VarLenType::BigVarBin =>
            {
                if let Some(bytes) = opt {
                    if bytes.len() > column_limit(vlc) {
                        return Err(crate::Error::BulkInput(
                            format!(
                                "Binary length {} exceed column limit {}",
                                bytes.len(),
                                column_limit(vlc)
                            )
                            .into(),
                        ));
                    }

                    if vlc.len() < 0xffff {
                        dst.put_u16_le(bytes.len() as u16);
                        dst.extend_from_slice(&bytes);
                    } else {
                        put_plp(dst, &bytes);
                    }
                } else if vlc.len() < 0xffff {
                    dst.put_u16_le(0xffff);
                } else {
                    dst.put_u64_le(0xffffffffffffffff);
                }
            }
            (ColumnData::Binary(Some(bytes)), None) if bytes.len() <= 8000 => {
                dst.put_u8(VarLenType::BigVarBin as u8);
                dst.put_u16_le(8000);
                dst.put_u16_le(bytes.len() as u16);
                dst.extend(bytes.into_owned());
            }
            (ColumnData::Binary(Some(bytes)), None) => put_varbinary_max(dst, Some(&bytes))?,
            (ColumnData::DateTime(Some(dt)), None) => {
                dst.extend_from_slice(&[VarLenType::Datetimen as u8, 8, 8]);
                dt.encode(&mut *dst)?;
            }
            (ColumnData::SmallDateTime(Some(dt)), None) => {
                dst.extend_from_slice(&[VarLenType::Datetimen as u8, 4, 4]);
                dt.encode(&mut *dst)?;
            }
            #[cfg(feature = "tds73")]
            (ColumnData::Date(opt), Some(TypeInfo::VarLenSized(vlc)))
                if vlc.r#type() == VarLenType::Daten =>
            {
                if let Some(dt) = opt {
                    dst.put_u8(3);
                    dt.encode(dst)?;
                } else {
                    dst.put_u8(0);
                }
            }
            #[cfg(feature = "tds73")]
            (ColumnData::Date(Some(date)), None) => {
                dst.extend_from_slice(&[VarLenType::Daten as u8, 3]);
                date.encode(&mut *dst)?;
            }
            #[cfg(feature = "tds73")]
            (ColumnData::Time(opt), Some(TypeInfo::VarLenSized(vlc)))
                if vlc.r#type() == VarLenType::Timen =>
            {
                if let Some(time) = opt {
                    let time = rescale_time(time, vlc.len() as u8)?;
                    dst.put_u8(time.len()?);
                    time.encode(dst)?;
                } else {
                    dst.put_u8(0);
                }
            }
            #[cfg(feature = "tds73")]
            (ColumnData::Time(Some(time)), None) => {
                dst.extend_from_slice(&[VarLenType::Timen as u8, time.scale(), time.len()?]);
                time.encode(&mut *dst)?;
            }
            #[cfg(feature = "tds73")]
            (ColumnData::DateTime2(opt), Some(TypeInfo::VarLenSized(vlc)))
                if vlc.r#type() == VarLenType::Datetime2 =>
            {
                if let Some(dt2) = opt {
                    let dt2 =
                        DateTime2::new(dt2.date(), rescale_time(dt2.time(), vlc.len() as u8)?);
                    dst.put_u8(dt2.time().len()? + 3);
                    dt2.encode(dst)?;
                } else {
                    dst.put_u8(0);
                }
            }
            #[cfg(feature = "tds73")]
            (ColumnData::DateTime2(Some(dt)), None) => {
                let len = dt.time().len()? + 3;
                dst.extend_from_slice(&[VarLenType::Datetime2 as u8, dt.time().scale(), len]);
                dt.encode(&mut *dst)?;
            }
            #[cfg(feature = "tds73")]
            (ColumnData::DateTimeOffset(opt), Some(TypeInfo::VarLenSized(vlc)))
                if vlc.r#type() == VarLenType::DatetimeOffsetn =>
            {
                if let Some(dto) = opt {
                    let dt2 = dto.datetime2();
                    let dt2 =
                        DateTime2::new(dt2.date(), rescale_time(dt2.time(), vlc.len() as u8)?);
                    let dto = DateTimeOffset::new(dt2, dto.offset());
                    dst.put_u8(dto.datetime2().time().len()? + 5);
                    dto.encode(dst)?;
                } else {
                    dst.put_u8(0);
                }
            }
            #[cfg(feature = "tds73")]
            (ColumnData::DateTimeOffset(Some(dto)), None) => {
                let headers = [
                    VarLenType::DatetimeOffsetn as u8,
                    dto.datetime2().time().scale(),
                    dto.datetime2().time().len()? + 5,
                ];

                dst.extend_from_slice(&headers);
                dto.encode(&mut *dst)?;
            }
            (ColumnData::Xml(opt), Some(TypeInfo::Xml { .. })) => {
                if let Some(xml) = opt {
                    xml.into_owned().encode(dst)?;
                } else {
                    dst.put_u64_le(0xffffffffffffffff_u64);
                }
            }
            // As SqlBulkCopy takes a string for xml; the column is declared
            // nvarchar(max), see `MetaDataColumn::encode`.
            (ColumnData::String(opt), Some(TypeInfo::Xml { .. })) => match opt {
                Some(s) => put_plp_utf16(dst, &s)?,
                None => dst.put_u64_le(PLP_NULL),
            },
            (ColumnData::Xml(Some(xml)), None) => {
                dst.put_u8(VarLenType::Xml as u8);
                dst.put_u8(0);
                xml.into_owned().encode(&mut *dst)?;
            }
            (ColumnData::Numeric(opt), Some(TypeInfo::VarLenSized(vlc)))
                if vlc.r#type() == VarLenType::Money =>
            {
                if let Some(num) = opt {
                    money::encode_numeric(dst, vlc.len(), num)?;
                } else {
                    dst.put_u8(0);
                }
            }
            (
                ColumnData::Numeric(opt),
                Some(TypeInfo::VarLenSizedPrecision {
                    ty,
                    precision,
                    scale,
                    ..
                }),
            ) if ty == &VarLenType::Numericn || ty == &VarLenType::Decimaln => {
                if let Some(num) = opt {
                    rescale_numeric(num, *precision, *scale)?.encode(&mut *dst)?;
                } else {
                    dst.put_u8(0);
                }
            }
            (ColumnData::Numeric(Some(num)), None) => {
                // Matches `type_name`. 17 is the maximum length for precision
                // 38; the value's own length byte may be shorter.
                let headers = &[
                    VarLenType::Numericn as u8,
                    17,
                    NUMERIC_PARAM_PRECISION,
                    num.scale(),
                ];

                dst.extend_from_slice(headers);
                num.encode(&mut *dst)?;
            }
            (data, Some(TypeInfo::VarLenSized(vlc))) if vlc.r#type() == VarLenType::SSVariant => {
                sql_variant::encode(&mut *dst, data, vlc.collation())?;
            }
            // Typed NULLs: an OUT parameter may not be an untyped NULL (8015).
            // Maximum precision and scale, so an OUT value loses nothing,
            // except numeric's scale 0, SqlClient's default for a NULL decimal.
            (ColumnData::String(None), None) => {
                dst.put_u8(VarLenType::NVarchar as u8);
                dst.put_u16_le(8000);
                dst.extend_from_slice(&[0u8; 5]);
                dst.put_u16_le(0xffff);
            }
            (ColumnData::Binary(None), None) => {
                dst.put_u8(VarLenType::BigVarBin as u8);
                dst.put_u16_le(8000);
                dst.put_u16_le(0xffff);
            }
            (ColumnData::Numeric(None), None) => {
                dst.extend_from_slice(&[
                    VarLenType::Numericn as u8,
                    17,
                    NUMERIC_PARAM_PRECISION,
                    0,
                    0,
                ]);
            }
            (ColumnData::Xml(None), None) => {
                dst.extend_from_slice(&[VarLenType::Xml as u8, 0]);
                dst.put_u64_le(PLP_NULL);
            }
            (ColumnData::DateTime(None), None) => {
                dst.extend_from_slice(&[VarLenType::Datetimen as u8, 8, 0]);
            }
            (ColumnData::SmallDateTime(None), None) => {
                dst.extend_from_slice(&[VarLenType::Datetimen as u8, 4, 0]);
            }
            #[cfg(feature = "tds73")]
            (ColumnData::Date(None), None) => {
                dst.extend_from_slice(&[VarLenType::Daten as u8, 0]);
            }
            #[cfg(feature = "tds73")]
            (ColumnData::Time(None), None) => {
                dst.extend_from_slice(&[VarLenType::Timen as u8, 7, 0]);
            }
            #[cfg(feature = "tds73")]
            (ColumnData::DateTime2(None), None) => {
                dst.extend_from_slice(&[VarLenType::Datetime2 as u8, 7, 0]);
            }
            #[cfg(feature = "tds73")]
            (ColumnData::DateTimeOffset(None), None) => {
                dst.extend_from_slice(&[VarLenType::DatetimeOffsetn as u8, 7, 0]);
            }
            (v, ref ti) => {
                return Err(crate::Error::BulkInput(
                    format!("invalid data type, expecting {:?} but found {:?}", ti, v).into(),
                ));
            }
        }

        Ok(())
    }
}

/// NOT NULL money is Moneyn without its length byte.
fn put_fixed_money(
    dst: &mut BytesMutWithTypeInfo<'_>,
    ty: FixedLenType,
    encode: impl FnOnce(&mut BytesMut, usize) -> crate::Result<()>,
) -> crate::Result<()> {
    let mut moneyn = BytesMut::new();
    encode(&mut moneyn, if ty == FixedLenType::Money4 { 4 } else { 8 })?;
    dst.extend_from_slice(&moneyn[1..]);
    Ok(())
}

/// A text, ntext or image value of `len` bytes, over the 2^31 - 1 they hold.
fn too_long(len: usize) -> crate::Error {
    crate::Error::BulkInput(
        format!("Encoded length {len} exceeds the column limit {}", i32::MAX).into(),
    )
}

/// The text pointer and timestamp SqlBulkCopy sends ahead of a text, ntext
/// or image value; the server ignores both.
fn put_text_pointer(dst: &mut BytesMutWithTypeInfo<'_>) {
    dst.put_u8(16);
    dst.extend_from_slice(&[0xff; 24]);
}

/// Reads a [`ColumnData`] straight out of a row without any conversion,
/// borrowing from the row's owned data.
impl<'a> FromSql<'a> for ColumnData<'a> {
    fn from_sql(value: &'a ColumnData<'static>) -> crate::Result<Option<Self>> {
        Ok(Some(value.clone()))
    }
}

/// Reads a [`ColumnData`] straight out of a row without any conversion, taking
/// ownership of the value.
impl<'a> FromSqlOwned for ColumnData<'a> {
    fn from_sql_owned(value: ColumnData<'static>) -> crate::Result<Option<Self>> {
        Ok(Some(value))
    }
}

/// Binds a [`ColumnData`] directly as a query parameter, by reference.
impl<'a> ToSql for ColumnData<'a> {
    fn to_sql(&self) -> ColumnData<'_> {
        self.clone()
    }
}

/// Binds a [`ColumnData`] directly as a query parameter, by value.
impl<'a> IntoSql<'a> for ColumnData<'a> {
    fn into_sql(self) -> ColumnData<'a> {
        self
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::sql_read_bytes::test_utils::IntoSqlReadBytes;
    use crate::tds::Collation;
    use crate::{Error, VarLenContext};
    use bytes::BytesMut;

    async fn test_round_trip(ti: TypeInfo, d: ColumnData<'_>) {
        let mut buf = BytesMut::new();
        let mut buf_with_ti = BytesMutWithTypeInfo::new(&mut buf).with_type_info(&ti);

        d.clone()
            .encode(&mut buf_with_ti)
            .expect("encode must succeed");

        let reader = &mut buf.into_sql_read_bytes();
        let nd = ColumnData::decode(reader, &ti)
            .await
            .expect("decode must succeed");

        assert_eq!(nd, d);

        reader
            .read_u8()
            .await
            .expect_err("decode must consume entire buffer");
    }

    #[test]
    fn type_name_maps_each_variant() {
        assert_eq!(ColumnData::U8(Some(1)).type_name(), "tinyint");
        assert_eq!(ColumnData::I16(Some(1)).type_name(), "smallint");
        assert_eq!(ColumnData::I32(Some(1)).type_name(), "int");
        assert_eq!(ColumnData::I64(Some(1)).type_name(), "bigint");
        assert_eq!(ColumnData::F32(Some(1.0)).type_name(), "float(24)");
        assert_eq!(ColumnData::F64(Some(1.0)).type_name(), "float(53)");
        assert_eq!(ColumnData::Bit(Some(true)).type_name(), "bit");
        assert_eq!(ColumnData::Guid(None).type_name(), "uniqueidentifier");
        assert_eq!(ColumnData::Numeric(None).type_name(), "numeric(38,0)");
        assert_eq!(ColumnData::DateTime(None).type_name(), "datetime");
        assert_eq!(ColumnData::SmallDateTime(None).type_name(), "smalldatetime");
    }

    #[test]
    fn type_name_string_length_thresholds() {
        // None and anything up to 4000 chars is a sized nvarchar; just past it
        // becomes nvarchar(max).
        assert_eq!(ColumnData::String(None).type_name(), "nvarchar(4000)");
        assert_eq!(
            ColumnData::String(Some("a".repeat(100).into())).type_name(),
            "nvarchar(4000)"
        );
        assert_eq!(
            ColumnData::String(Some("a".repeat(4000).into())).type_name(),
            "nvarchar(4000)"
        );
        assert_eq!(
            ColumnData::String(Some("a".repeat(4001).into())).type_name(),
            "nvarchar(max)"
        );
    }

    /// Decodes what `encode_rpc_param` wrote: TYPE_INFO, then the value.
    async fn decode_rpc_param(buf: BytesMut) -> (TypeInfo, ColumnData<'static>) {
        let reader = &mut buf.into_sql_read_bytes();
        let ti = TypeInfo::decode(reader).await.unwrap();
        let data = ColumnData::decode(reader, &ti).await.unwrap();
        reader
            .read_u8()
            .await
            .expect_err("decode must consume entire buffer");
        (ti, data)
    }

    #[tokio::test]
    async fn rpc_nulls_are_typed() {
        let nulls = vec![
            ColumnData::U8(None),
            ColumnData::I16(None),
            ColumnData::I32(None),
            ColumnData::I64(None),
            ColumnData::F32(None),
            ColumnData::F64(None),
            ColumnData::Bit(None),
            ColumnData::String(None),
            ColumnData::Guid(None),
            ColumnData::Binary(None),
            ColumnData::Numeric(None),
            ColumnData::Xml(None),
            ColumnData::DateTime(None),
            ColumnData::SmallDateTime(None),
            #[cfg(feature = "tds73")]
            ColumnData::Time(None),
            #[cfg(feature = "tds73")]
            ColumnData::Date(None),
            #[cfg(feature = "tds73")]
            ColumnData::DateTime2(None),
            #[cfg(feature = "tds73")]
            ColumnData::DateTimeOffset(None),
        ];

        for null in nulls {
            for out in [false, true] {
                let mut buf = BytesMut::new();
                null.clone().encode_rpc_param(&mut buf, out, None).unwrap();
                assert_ne!(buf[0], FixedLenType::Null as u8, "{null:?}");

                let (_, decoded) = decode_rpc_param(buf).await;
                assert_eq!(decoded, null);
            }
        }
    }

    #[tokio::test]
    async fn varchar_params_go_in_the_session_collation() {
        // Windows-1252, as SQL_Latin1_General_CP1_CI_AS.
        let collation = Some(Collation::new(0x0409, 0));
        let text = |v: &ColumnData<'_>| match v {
            ColumnData::VarChar(s) | ColumnData::String(s) => s.clone().map(Cow::into_owned),
            _ => None,
        };

        for (value, out, expected_len) in [
            (ColumnData::VarChar(Some("Väinämöinen".into())), false, 8000),
            (
                ColumnData::VarChar(Some("Väinämöinen".into())),
                true,
                0xffff,
            ),
            (ColumnData::VarChar(None), false, 8000),
            (ColumnData::VarChar(None), true, 0xffff),
            (
                ColumnData::VarChar(Some("a".repeat(8001).into())),
                false,
                0xffff,
            ),
        ] {
            let mut buf = BytesMut::new();
            value
                .clone()
                .encode_rpc_param(&mut buf, out, collation)
                .unwrap();

            // One byte per character, so the TYPE_INFO, the 2-byte length and
            // 11 bytes of value, when not PLP.
            if expected_len == 8000 {
                let value_len = text(&value).map_or(0, |s| s.chars().count());
                assert_eq!(buf.len(), 1 + 2 + 5 + 2 + value_len, "{value:?}");
            }

            let (ti, decoded) = decode_rpc_param(buf).await;
            match ti {
                TypeInfo::VarLenSized(vlc) => {
                    assert_eq!(vlc.r#type(), VarLenType::BigVarChar, "{value:?}");
                    assert_eq!(vlc.len(), expected_len, "{value:?}");
                    assert_eq!(vlc.collation(), collation, "{value:?}");
                }
                other => panic!("{other:?}"),
            }
            // Read back as the plain string every character column is.
            assert_eq!(decoded, ColumnData::String(text(&value).map(Cow::Owned)));
        }
    }

    #[test]
    fn a_varchar_param_needs_a_collation_and_a_character_it_holds() {
        let mut buf = BytesMut::new();
        let err = ColumnData::VarChar(Some("a".into()))
            .encode_rpc_param(&mut buf, false, None)
            .unwrap_err();
        assert!(matches!(err, Error::Encoding(_)), "{err:?}");

        let collation = Some(Collation::new(0x0409, 0));
        let err = ColumnData::VarChar(Some("日本".into()))
            .encode_rpc_param(&mut buf, false, collation)
            .unwrap_err();
        assert!(matches!(err, Error::Encoding(_)), "{err:?}");
        assert!(buf.is_empty(), "nothing written before the error");
    }

    #[test]
    fn varchar_params_are_declared_by_utf8_length() {
        assert_eq!(ColumnData::VarChar(None).type_name(), "varchar(8000)");
        assert_eq!(
            ColumnData::VarChar(Some("a".repeat(8000).into())).type_name(),
            "varchar(8000)"
        );
        assert_eq!(
            ColumnData::VarChar(Some("a".repeat(8001).into())).type_name(),
            "varchar(max)"
        );
    }

    #[tokio::test]
    async fn out_strings_and_binaries_are_sent_as_max() {
        let values = [
            ColumnData::String(Some("a".into())),
            ColumnData::String(None),
            ColumnData::Binary(Some(vec![1u8].into())),
            ColumnData::Binary(None),
        ];

        for value in values {
            for (out, max_len) in [(true, 0xffff), (false, 8000)] {
                let mut buf = BytesMut::new();
                value.clone().encode_rpc_param(&mut buf, out, None).unwrap();

                let (ti, decoded) = decode_rpc_param(buf).await;
                match ti {
                    TypeInfo::VarLenSized(vlc) => assert_eq!(vlc.len(), max_len, "{value:?}"),
                    other => panic!("{other:?}"),
                }
                assert_eq!(decoded, value);
            }
        }
    }

    #[tokio::test]
    async fn nulls_are_declared_like_values_of_their_type() {
        // A declaration of its own gave a NULL its own cached plan. The value
        // is short, and numeric's at scale 0, the scale SqlClient gives a NULL.
        let pairs = vec![
            (
                ColumnData::Binary(None),
                ColumnData::Binary(Some(vec![1].into())),
            ),
            (
                ColumnData::Numeric(None),
                ColumnData::Numeric(Some(Numeric::new_with_scale(5, 0))),
            ),
            (
                ColumnData::String(None),
                ColumnData::String(Some("a".into())),
            ),
            (ColumnData::U8(None), ColumnData::U8(Some(1))),
            (ColumnData::I16(None), ColumnData::I16(Some(1))),
            (ColumnData::I32(None), ColumnData::I32(Some(1))),
            (ColumnData::I64(None), ColumnData::I64(Some(1))),
            (ColumnData::F32(None), ColumnData::F32(Some(1.0))),
            (ColumnData::F64(None), ColumnData::F64(Some(1.0))),
            (ColumnData::Bit(None), ColumnData::Bit(Some(true))),
            (ColumnData::Guid(None), ColumnData::Guid(Some(Uuid::nil()))),
            (
                ColumnData::Xml(None),
                ColumnData::Xml(Some(Cow::Owned(XmlData::new("<a/>")))),
            ),
            (
                ColumnData::DateTime(None),
                ColumnData::DateTime(Some(DateTime::new(0, 0))),
            ),
            (
                ColumnData::SmallDateTime(None),
                ColumnData::SmallDateTime(Some(SmallDateTime::new(0, 0))),
            ),
            #[cfg(feature = "tds73")]
            (
                ColumnData::Time(None),
                ColumnData::Time(Some(Time::new(0, 7))),
            ),
            #[cfg(feature = "tds73")]
            (ColumnData::Date(None), ColumnData::Date(Some(Date::new(0)))),
            #[cfg(feature = "tds73")]
            (
                ColumnData::DateTime2(None),
                ColumnData::DateTime2(Some(DateTime2::new(Date::new(0), Time::new(0, 7)))),
            ),
            #[cfg(feature = "tds73")]
            (
                ColumnData::DateTimeOffset(None),
                ColumnData::DateTimeOffset(Some(DateTimeOffset::new(
                    DateTime2::new(Date::new(0), Time::new(0, 7)),
                    0,
                ))),
            ),
        ];

        for (null, value) in pairs {
            assert_eq!(null.type_name(), value.type_name(), "{null:?}");

            // And the TYPE_INFO sent with the NULL is the declared type's.
            let ti = |data: ColumnData<'static>| async {
                let mut buf = BytesMut::new();
                data.encode_rpc_param(&mut buf, false, None).unwrap();
                format!("{:?}", decode_rpc_param(buf).await.0)
            };
            assert_eq!(ti(null.clone()).await, ti(value).await, "{null:?}");
        }
    }

    #[test]
    fn numeric_parameters_are_declared_at_precision_38() {
        for (value, scale) in [
            (15i128, 1u8),
            (123_455, 1),
            (10i128.pow(38) - 1, 38),
            (-7, 0),
        ] {
            let data = ColumnData::Numeric(Some(Numeric::new_with_scale(value, scale)));
            assert_eq!(data.type_name(), format!("numeric(38,{scale})"));

            // TYPE_INFO: type, max length, precision, scale.
            let mut buf = BytesMut::new();
            data.encode(&mut BytesMutWithTypeInfo::new(&mut buf))
                .unwrap();
            assert_eq!(&buf[..4], &[VarLenType::Numericn as u8, 17, 38, scale]);
        }
    }

    #[test]
    fn nvarchar_4000_threshold_counts_utf16_units() {
        // (string, fits nvarchar(4000)); € is 3 UTF-8 bytes and 1 unit, 😀 is
        // 4 bytes and 2 units.
        let cases = [
            ("\u{20AC}".repeat(1500), true),
            ("\u{20AC}".repeat(4000), true),
            ("\u{20AC}".repeat(4001), false),
            ("\u{1F600}".repeat(2000), true),
            ("\u{1F600}".repeat(2001), false),
        ];

        for (s, fits) in cases {
            let data = ColumnData::String(Some(s.into()));
            let expected = if fits {
                "nvarchar(4000)"
            } else {
                "nvarchar(max)"
            };
            assert_eq!(data.type_name(), expected);

            // The encoder must agree: TYPE_INFO max length 8000, or 0xFFFF for (max).
            let mut buf = BytesMut::new();
            data.encode(&mut BytesMutWithTypeInfo::new(&mut buf))
                .unwrap();
            assert_eq!(buf[0], VarLenType::NVarchar as u8);
            let max_len = u16::from_le_bytes([buf[1], buf[2]]);
            assert_eq!(max_len, if fits { 8000 } else { 0xFFFF }, "{expected}");
        }
    }

    /// 3 UTF-8 bytes, 1 UTF-16 unit each: over 2^30 bytes, so once declared
    /// ntext(max), but within the 2^30 - 1 units nvarchar(max) holds.
    fn euros_over_1_gib() -> String {
        "\u{20AC}".repeat((1 << 30) / 3 + 1)
    }

    #[test]
    fn strings_over_1_gib_of_utf8() {
        let euros = euros_over_1_gib();
        assert_eq!(
            ColumnData::String(Some(euros.into())).type_name(),
            "nvarchar(max)"
        );

        // 2^30 units, one more than nvarchar(max) holds.
        let err = ColumnData::String(Some("a".repeat(1 << 30).into()))
            .encode_rpc_param(&mut BytesMut::new(), false, None)
            .unwrap_err();
        assert!(matches!(err, Error::Conversion(_)), "{err:?}");
    }

    #[test]
    #[ignore = "writes 716 MB, about 40 s in a debug build"]
    fn string_over_1_gib_of_utf8_encodes() {
        let euros = euros_over_1_gib();
        let units = euros.len() / 3;

        let mut buf = BytesMut::new();
        ColumnData::String(Some(euros.into()))
            .encode_rpc_param(&mut buf, false, None)
            .unwrap();

        // TYPE_INFO (8 bytes), unknown length (8), chunk length, units, terminator.
        assert_eq!(buf[..3], [VarLenType::NVarchar as u8, 0xff, 0xff]);
        let chunk = u32::from_le_bytes(buf[16..20].try_into().unwrap());
        assert_eq!(chunk as usize, units * 2);
        assert_eq!(buf.len(), 20 + units * 2 + 4);
    }

    #[test]
    fn type_name_binary_length_threshold() {
        assert_eq!(
            ColumnData::Binary(Some(vec![0u8; 8000].into())).type_name(),
            "varbinary(8000)"
        );
        assert_eq!(
            ColumnData::Binary(Some(vec![0u8; 8001].into())).type_name(),
            "varbinary(max)"
        );
        assert_eq!(ColumnData::Binary(None).type_name(), "varbinary(8000)");
    }

    #[test]
    #[cfg(feature = "serde")]
    fn serde_json_round_trip() {
        let value = ColumnData::I32(Some(1234));
        let json = serde_json::to_string(&value).expect("serialize");
        let back: ColumnData<'static> = serde_json::from_str(&json).expect("deserialize");
        assert_eq!(value, back);
    }

    #[tokio::test]
    async fn i32_with_varlen_int() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Intn, 4, None)),
            ColumnData::I32(Some(42)),
        )
        .await;
    }

    #[tokio::test]
    async fn none_with_varlen_int() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Intn, 4, None)),
            ColumnData::I32(None),
        )
        .await;
    }

    #[tokio::test]
    async fn i32_with_fixedlen_int() {
        test_round_trip(
            TypeInfo::FixedLen(FixedLenType::Int4),
            ColumnData::I32(Some(42)),
        )
        .await;
    }

    #[tokio::test]
    async fn bit_with_varlen_bit() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Bitn, 1, None)),
            ColumnData::Bit(Some(true)),
        )
        .await;
    }

    #[tokio::test]
    async fn none_with_varlen_bit() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Bitn, 1, None)),
            ColumnData::Bit(None),
        )
        .await;
    }

    #[tokio::test]
    async fn bit_with_fixedlen_bit() {
        test_round_trip(
            TypeInfo::FixedLen(FixedLenType::Bit),
            ColumnData::Bit(Some(true)),
        )
        .await;
    }

    #[tokio::test]
    async fn u8_with_varlen_int() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Intn, 1, None)),
            ColumnData::U8(Some(8u8)),
        )
        .await;
    }

    #[tokio::test]
    async fn none_u8_with_varlen_int() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Intn, 1, None)),
            ColumnData::U8(None),
        )
        .await;
    }

    #[tokio::test]
    async fn u8_with_fixedlen_int() {
        test_round_trip(
            TypeInfo::FixedLen(FixedLenType::Int1),
            ColumnData::U8(Some(8u8)),
        )
        .await;
    }

    #[tokio::test]
    async fn i16_with_varlen_intn() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Intn, 2, None)),
            ColumnData::I16(Some(8i16)),
        )
        .await;
    }

    #[tokio::test]
    async fn none_i16_with_varlen_intn() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Intn, 2, None)),
            ColumnData::I16(None),
        )
        .await;
    }

    #[tokio::test]
    async fn none_with_varlen_intn() {
        test_round_trip(
            TypeInfo::FixedLen(FixedLenType::Int2),
            ColumnData::I16(Some(8i16)),
        )
        .await;
    }

    #[tokio::test]
    async fn i64_with_varlen_intn() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Intn, 8, None)),
            ColumnData::I64(Some(8i64)),
        )
        .await;
    }

    #[tokio::test]
    async fn i64_none_with_varlen_intn() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Intn, 8, None)),
            ColumnData::I64(None),
        )
        .await;
    }

    #[tokio::test]
    async fn i64_with_fixedlen_int8() {
        test_round_trip(
            TypeInfo::FixedLen(FixedLenType::Int8),
            ColumnData::I64(Some(8i64)),
        )
        .await;
    }

    #[tokio::test]
    async fn f32_with_varlen_floatn() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Floatn, 4, None)),
            ColumnData::F32(Some(8f32)),
        )
        .await;
    }

    #[tokio::test]
    async fn null_f32_with_varlen_floatn() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Floatn, 4, None)),
            ColumnData::F32(None),
        )
        .await;
    }

    #[tokio::test]
    async fn f32_with_fixedlen_float4() {
        test_round_trip(
            TypeInfo::FixedLen(FixedLenType::Float4),
            ColumnData::F32(Some(8f32)),
        )
        .await;
    }

    #[tokio::test]
    async fn f64_with_varlen_floatn() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Floatn, 8, None)),
            ColumnData::F64(Some(8f64)),
        )
        .await;
    }

    #[tokio::test]
    async fn none_f64_with_varlen_floatn() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Floatn, 8, None)),
            ColumnData::F64(None),
        )
        .await;
    }

    #[tokio::test]
    async fn f64_with_fixedlen_float8() {
        test_round_trip(
            TypeInfo::FixedLen(FixedLenType::Float8),
            ColumnData::F64(Some(8f64)),
        )
        .await;
    }

    #[tokio::test]
    async fn guid_with_varlen_guid() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Guid, 16, None)),
            ColumnData::Guid(Some(Uuid::new_v4())),
        )
        .await;
    }

    #[tokio::test]
    async fn none_guid_with_varlen_guid() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Guid, 16, None)),
            ColumnData::Guid(None),
        )
        .await;
    }

    #[tokio::test]
    async fn numeric_with_varlen_sized_precision() {
        test_round_trip(
            TypeInfo::VarLenSizedPrecision {
                ty: VarLenType::Numericn,
                size: 17,
                precision: 18,
                scale: 0,
            },
            ColumnData::Numeric(Some(Numeric::new_with_scale(23, 0))),
        )
        .await;
    }

    #[tokio::test]
    async fn none_numeric_with_varlen_sized_precision() {
        test_round_trip(
            TypeInfo::VarLenSizedPrecision {
                ty: VarLenType::Numericn,
                size: 17,
                precision: 18,
                scale: 0,
            },
            ColumnData::Numeric(None),
        )
        .await;
    }

    #[tokio::test]
    async fn string_with_varlen_bigchar() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(
                VarLenType::BigChar,
                40,
                Some(Collation::new(13632521, 52)),
            )),
            ColumnData::String(Some("aaa".into())),
        )
        .await;
    }

    #[tokio::test]
    async fn long_string_with_varlen_bigchar() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(
                VarLenType::BigChar,
                0x8ffff,
                Some(Collation::new(13632521, 52)),
            )),
            ColumnData::String(Some("aaa".into())),
        )
        .await;
    }

    #[tokio::test]
    async fn none_long_string_with_varlen_bigchar() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(
                VarLenType::BigChar,
                0x8ffff,
                Some(Collation::new(13632521, 52)),
            )),
            ColumnData::String(None),
        )
        .await;
    }

    #[tokio::test]
    async fn none_string_with_varlen_bigchar() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(
                VarLenType::BigChar,
                40,
                Some(Collation::new(13632521, 52)),
            )),
            ColumnData::String(None),
        )
        .await;
    }

    #[tokio::test]
    async fn string_with_varlen_bigvarchar() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(
                VarLenType::BigVarChar,
                40,
                Some(Collation::new(13632521, 52)),
            )),
            ColumnData::String(Some("aaa".into())),
        )
        .await;
    }

    #[tokio::test]
    async fn none_string_with_varlen_bigvarchar() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(
                VarLenType::BigVarChar,
                40,
                Some(Collation::new(13632521, 52)),
            )),
            ColumnData::String(None),
        )
        .await;
    }

    #[tokio::test]
    async fn empty_string_with_varlen_bigvarchar() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(
                VarLenType::BigVarChar,
                0x8ffff,
                Some(Collation::new(13632521, 52)),
            )),
            ColumnData::String(Some("".into())),
        )
        .await;
    }

    #[tokio::test]
    async fn string_with_varlen_nvarchar() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(
                VarLenType::NVarchar,
                40,
                Some(Collation::new(13632521, 52)),
            )),
            ColumnData::String(Some("hhh".into())),
        )
        .await;
    }

    #[tokio::test]
    async fn none_string_with_varlen_nvarchar() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(
                VarLenType::NVarchar,
                40,
                Some(Collation::new(13632521, 52)),
            )),
            ColumnData::String(None),
        )
        .await;
    }

    #[tokio::test]
    async fn empty_string_with_varlen_nvarchar() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(
                VarLenType::NVarchar,
                0x8ffff,
                Some(Collation::new(13632521, 52)),
            )),
            ColumnData::String(Some("".into())),
        )
        .await;
    }

    #[tokio::test]
    async fn string_with_varlen_nchar() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(
                VarLenType::NChar,
                40,
                Some(Collation::new(13632521, 52)),
            )),
            ColumnData::String(Some("hhh".into())),
        )
        .await;
    }

    #[tokio::test]
    async fn long_string_with_varlen_nchar() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(
                VarLenType::NChar,
                0x8ffff,
                Some(Collation::new(13632521, 52)),
            )),
            ColumnData::String(Some("hhh".into())),
        )
        .await;
    }

    #[tokio::test]
    async fn none_long_string_with_varlen_nchar() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(
                VarLenType::NChar,
                0x8ffff,
                Some(Collation::new(13632521, 52)),
            )),
            ColumnData::String(None),
        )
        .await;
    }

    #[tokio::test]
    async fn none_string_with_varlen_nchar() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(
                VarLenType::NChar,
                40,
                Some(Collation::new(13632521, 52)),
            )),
            ColumnData::String(None),
        )
        .await;
    }

    #[tokio::test]
    async fn binary_with_varlen_bigbinary() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(VarLenType::BigBinary, 40, None)),
            ColumnData::Binary(Some(b"aaa".as_slice().into())),
        )
        .await;
    }

    #[tokio::test]
    async fn long_binary_with_varlen_bigbinary() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(VarLenType::BigBinary, 0x8ffff, None)),
            ColumnData::Binary(Some(b"aaa".as_slice().into())),
        )
        .await;
    }

    #[tokio::test]
    async fn none_binary_with_varlen_bigbinary() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(VarLenType::BigBinary, 40, None)),
            ColumnData::Binary(None),
        )
        .await;
    }

    #[tokio::test]
    async fn none_long_binary_with_varlen_bigbinary() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(VarLenType::BigBinary, 0x8ffff, None)),
            ColumnData::Binary(None),
        )
        .await;
    }

    #[tokio::test]
    async fn binary_with_varlen_bigvarbin() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(VarLenType::BigVarBin, 40, None)),
            ColumnData::Binary(Some(b"aaa".as_slice().into())),
        )
        .await;
    }

    #[tokio::test]
    async fn none_binary_with_varlen_bigvarbin() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(VarLenType::BigVarBin, 40, None)),
            ColumnData::Binary(None),
        )
        .await;
    }

    #[tokio::test]
    async fn empty_binary_with_varlen_bigvarbin() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(
                VarLenType::BigVarBin,
                0x8ffff,
                Some(Collation::new(13632521, 52)),
            )),
            ColumnData::Binary(Some(b"".as_slice().into())),
        )
        .await;
    }

    #[tokio::test]
    async fn datetime_with_varlen_datetimen() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Datetimen, 8, None)),
            ColumnData::DateTime(Some(DateTime::new(200, 3000))),
        )
        .await;
    }

    // this is inconsistent: decode will decode any None datetime to smalldatetime, ignoring size
    // but it's non-critical, so let it be here
    #[tokio::test]
    async fn none_datetime_with_varlen_datetimen() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Datetimen, 8, None)),
            ColumnData::DateTime(None),
        )
        .await;
    }

    #[tokio::test]
    async fn datetime_with_fixedlen_datetime() {
        test_round_trip(
            TypeInfo::FixedLen(FixedLenType::Datetime),
            ColumnData::DateTime(Some(DateTime::new(200, 3000))),
        )
        .await;
    }

    #[tokio::test]
    async fn smalldatetime_with_varlen_datetimen() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Datetimen, 4, None)),
            ColumnData::SmallDateTime(Some(SmallDateTime::new(200, 1439))),
        )
        .await;
    }

    #[tokio::test]
    async fn none_smalldatetime_with_varlen_datetimen() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Datetimen, 4, None)),
            ColumnData::SmallDateTime(None),
        )
        .await;
    }

    #[tokio::test]
    async fn smalldatetime_with_fixedlen_datetime4() {
        test_round_trip(
            TypeInfo::FixedLen(FixedLenType::Datetime4),
            ColumnData::SmallDateTime(Some(SmallDateTime::new(200, 1439))),
        )
        .await;
    }

    #[cfg(feature = "tds73")]
    #[tokio::test]
    async fn date_with_varlen_daten() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Daten, 3, None)),
            ColumnData::Date(Some(Date::new(200))),
        )
        .await;
    }

    #[cfg(feature = "tds73")]
    #[tokio::test]
    async fn none_date_with_varlen_daten() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Daten, 3, None)),
            ColumnData::Date(None),
        )
        .await;
    }

    #[cfg(feature = "tds73")]
    #[tokio::test]
    async fn time_with_varlen_timen() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Timen, 7, None)),
            ColumnData::Time(Some(Time::new(55, 7))),
        )
        .await;
    }

    #[cfg(feature = "tds73")]
    #[tokio::test]
    async fn none_time_with_varlen_timen() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Timen, 7, None)),
            ColumnData::Time(None),
        )
        .await;
    }

    #[cfg(feature = "tds73")]
    #[tokio::test]
    async fn datetime2_with_varlen_datetime2() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Datetime2, 7, None)),
            ColumnData::DateTime2(Some(DateTime2::new(Date::new(55), Time::new(222, 7)))),
        )
        .await;
    }

    #[cfg(feature = "tds73")]
    #[tokio::test]
    async fn none_datetime2_with_varlen_datetime2() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Datetime2, 7, None)),
            ColumnData::DateTime2(None),
        )
        .await;
    }

    #[cfg(feature = "tds73")]
    #[tokio::test]
    async fn datetimeoffset_with_varlen_datetimeoffsetn() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(VarLenType::DatetimeOffsetn, 7, None)),
            ColumnData::DateTimeOffset(Some(DateTimeOffset::new(
                DateTime2::new(Date::new(55), Time::new(222, 7)),
                -8,
            ))),
        )
        .await;
    }

    #[cfg(feature = "tds73")]
    #[tokio::test]
    async fn none_datetimeoffset_with_varlen_datetimeoffsetn() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(VarLenType::DatetimeOffsetn, 7, None)),
            ColumnData::DateTimeOffset(None),
        )
        .await;
    }

    #[cfg(feature = "tds73")]
    #[tokio::test]
    async fn xml_with_xml() {
        test_round_trip(
            TypeInfo::Xml {
                schema: None,
                size: 0xfffffffffffffffe_usize,
            },
            ColumnData::Xml(Some(Cow::Owned(XmlData::new("<a>ddd</a>")))),
        )
        .await;
    }

    #[cfg(feature = "tds73")]
    #[tokio::test]
    async fn none_xml_with_xml() {
        test_round_trip(
            TypeInfo::Xml {
                schema: None,
                size: 0xfffffffffffffffe_usize,
            },
            ColumnData::Xml(None),
        )
        .await;
    }

    #[tokio::test]
    async fn invalid_type_fails() {
        let data = vec![
            (
                TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Floatn, 4, None)),
                ColumnData::I32(Some(42)),
            ),
            (
                TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Floatn, 4, None)),
                ColumnData::I32(None),
            ),
            (
                TypeInfo::FixedLen(FixedLenType::Int4),
                ColumnData::I32(None),
            ),
        ];

        for (ti, d) in data {
            let mut buf = BytesMut::new();
            let mut buf_ti = BytesMutWithTypeInfo::new(&mut buf).with_type_info(&ti);

            let err = d.encode(&mut buf_ti).expect_err("encode should fail");

            if let Error::BulkInput(_) = err {
            } else {
                panic!("Expected: Error::BulkInput, got: {:?}", err);
            }
        }
    }

    #[test]
    fn column_data_from_sql_clones_by_reference() {
        let value: ColumnData<'static> = ColumnData::I32(Some(42));
        let out = ColumnData::from_sql(&value).unwrap();
        assert_eq!(out, Some(ColumnData::I32(Some(42))));
    }

    #[test]
    fn column_data_from_sql_owned_passes_through() {
        let value: ColumnData<'static> = ColumnData::String(Some(Cow::Borrowed("hello")));
        let out = ColumnData::from_sql_owned(value).unwrap();
        assert_eq!(out, Some(ColumnData::String(Some(Cow::Borrowed("hello")))));
    }

    #[test]
    fn column_data_to_sql_clones_by_reference() {
        let value = ColumnData::Bit(Some(true));
        assert_eq!(value.to_sql(), ColumnData::Bit(Some(true)));
        // Original is untouched.
        assert_eq!(value, ColumnData::Bit(Some(true)));
    }

    #[test]
    fn column_data_into_sql_passes_through() {
        let value = ColumnData::F64(Some(1.5));
        assert_eq!(value.into_sql(), ColumnData::F64(Some(1.5)));
    }

    // ----- helpers for the coverage tests below -----

    fn encode_with_ti(ti: &TypeInfo, d: ColumnData<'_>) -> crate::Result<BytesMut> {
        let mut buf = BytesMut::new();
        {
            let mut b = BytesMutWithTypeInfo::new(&mut buf).with_type_info(ti);
            d.encode(&mut b)?;
        }
        Ok(buf)
    }

    fn encode_without_ti(d: ColumnData<'_>) -> crate::Result<BytesMut> {
        let mut buf = BytesMut::new();
        {
            let mut b = BytesMutWithTypeInfo::new(&mut buf);
            d.encode(&mut b)?;
        }
        Ok(buf)
    }

    fn expect_bulk_input(ti: TypeInfo, d: ColumnData<'_>) {
        let mut buf = BytesMut::new();
        let mut b = BytesMutWithTypeInfo::new(&mut buf).with_type_info(&ti);
        let err = d.encode(&mut b).expect_err("encode should fail");
        assert!(matches!(err, Error::BulkInput(_)), "got {:?}", err);
    }

    #[tokio::test]
    async fn decode_varlen_sized_precision_non_decimal_is_an_error() {
        let ti = TypeInfo::VarLenSizedPrecision {
            ty: VarLenType::Money,
            size: 8,
            precision: 0,
            scale: 0,
        };
        let buf = BytesMut::new();
        let reader = &mut buf.into_sql_read_bytes();
        let res = ColumnData::decode(reader, &ti).await;
        assert!(matches!(res, Err(Error::Protocol(_))), "{res:?}");
    }

    // ----- decode: line 202 (Udt) -----

    #[tokio::test]
    async fn decode_udt_type_info() {
        use bytes::BufMut;

        let ti = TypeInfo::Udt(crate::tds::codec::type_info::UdtInfo {
            max_byte_size: 0xffff,
            db_name: "db".into(),
            schema_name: "dbo".into(),
            type_name: "geometry".into(),
            assembly_qualified_name: String::new(),
        });

        let mut buf = BytesMut::new();
        // PLP unknown-length sentinel + one chunk + terminator.
        buf.put_u64_le(0xfffffffffffffffe);
        buf.put_u32_le(4);
        buf.extend_from_slice(&[1, 2, 3, 4]);
        buf.put_u32_le(0);

        let reader = &mut buf.into_sql_read_bytes();
        let nd = ColumnData::decode(reader, &ti).await.unwrap();
        assert_eq!(nd, ColumnData::Binary(Some(vec![1, 2, 3, 4].into())));
    }

    // ----- F64 with Money (lines 376-383) -----

    #[tokio::test]
    async fn f64_with_varlen_money() {
        let ti = TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Money, 8, None));
        let buf = encode_with_ti(&ti, ColumnData::F64(Some(3.5))).unwrap();

        // Money decodes exactly, to Numeric.
        let nd = ColumnData::decode(&mut buf.into_sql_read_bytes(), &ti)
            .await
            .unwrap();
        assert_eq!(
            nd,
            ColumnData::Numeric(Some(Numeric::new_with_scale(35000, 4)))
        );
    }

    #[tokio::test]
    async fn none_f64_with_varlen_money() {
        let ti = TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Money, 8, None));
        let buf = encode_with_ti(&ti, ColumnData::F64(None)).unwrap();

        let nd = ColumnData::decode(&mut buf.into_sql_read_bytes(), &ti)
            .await
            .unwrap();
        assert_eq!(nd, ColumnData::Numeric(None));
    }

    // ----- BigChar error paths (lines 426, 430-437) -----

    #[tokio::test]
    async fn bigchar_unrepresentable_character_errors() {
        let ti = TypeInfo::VarLenSized(VarLenContext::new(
            VarLenType::BigChar,
            40,
            Some(Collation::new(13632521, 52)),
        ));
        let mut buf = BytesMut::new();
        let mut b = BytesMutWithTypeInfo::new(&mut buf).with_type_info(&ti);
        // An emoji has no representation in the (single-byte) column collation.
        let err = ColumnData::String(Some("\u{1F600}".into()))
            .encode(&mut b)
            .expect_err("encode should fail");
        assert!(matches!(err, Error::Encoding(_)), "got {:?}", err);
    }

    #[tokio::test]
    async fn bigchar_too_long_errors() {
        expect_bulk_input(
            TypeInfo::VarLenSized(VarLenContext::new(
                VarLenType::BigChar,
                2,
                Some(Collation::new(13632521, 52)),
            )),
            ColumnData::String(Some("aaa".into())),
        );
    }

    // ----- NVarchar error paths (lines 481-488 small, 513-520 unknown-size) -----

    #[tokio::test]
    async fn nvarchar_too_long_small_errors() {
        expect_bulk_input(
            TypeInfo::VarLenSized(VarLenContext::new(
                VarLenType::NVarchar,
                2,
                Some(Collation::new(13632521, 52)),
            )),
            ColumnData::String(Some("aaa".into())),
        );
    }

    // ----- Text / NText encode arms (lines 538-587) -----

    #[tokio::test]
    async fn string_with_varlen_text() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(
                VarLenType::Text,
                40,
                Some(Collation::new(13632521, 52)),
            )),
            ColumnData::String(Some("hello".into())),
        )
        .await;
    }

    #[tokio::test]
    async fn none_string_with_varlen_text() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(
                VarLenType::Text,
                40,
                Some(Collation::new(13632521, 52)),
            )),
            ColumnData::String(None),
        )
        .await;
    }

    #[tokio::test]
    async fn string_with_varlen_ntext() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(VarLenType::NText, 40, None)),
            ColumnData::String(Some("hi".into())),
        )
        .await;
    }

    #[tokio::test]
    async fn none_string_with_varlen_ntext() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(VarLenType::NText, 40, None)),
            ColumnData::String(None),
        )
        .await;
    }

    // ----- Binary too long (lines 650-657) -----

    #[tokio::test]
    async fn binary_too_long_errors() {
        expect_bulk_input(
            TypeInfo::VarLenSized(VarLenContext::new(VarLenType::BigVarBin, 2, None)),
            ColumnData::Binary(Some(b"aaa".as_slice().into())),
        );
    }

    // ----- DateTime / SmallDateTime encode without TypeInfo (714-716, 734-736) -----

    #[tokio::test]
    async fn datetime_encode_without_type_info() {
        let buf = encode_without_ti(ColumnData::DateTime(Some(DateTime::new(200, 3000)))).unwrap();
        assert_eq!(buf[0], VarLenType::Datetimen as u8);
        assert_eq!(buf[1], 8);
        assert_eq!(buf[2], 8);
    }

    #[tokio::test]
    async fn smalldatetime_encode_without_type_info() {
        let buf = encode_without_ti(ColumnData::SmallDateTime(Some(SmallDateTime::new(
            200, 600,
        ))))
        .unwrap();
        assert_eq!(buf[0], VarLenType::Datetimen as u8);
        assert_eq!(buf[1], 4);
        assert_eq!(buf[2], 4);
    }

    // ----- DateTime2 into a `datetime` (Datetimen) column (lines 774-780) -----

    #[cfg(feature = "tds73")]
    #[tokio::test]
    async fn datetime2_with_varlen_datetimen() {
        let ti = TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Datetimen, 8, None));
        // 2020-01-01 is representable by `datetime` (>= 1900-01-01).
        let dt2 = DateTime2::new(Date::new(737_425), Time::new(0, 7));
        let buf = encode_with_ti(&ti, ColumnData::DateTime2(Some(dt2))).unwrap();

        let reader = &mut buf.into_sql_read_bytes();
        let nd = ColumnData::decode(reader, &ti).await.unwrap();
        assert!(matches!(nd, ColumnData::DateTime(Some(_))), "got {:?}", nd);
    }

    #[cfg(feature = "tds73")]
    #[tokio::test]
    async fn none_datetime2_with_varlen_datetimen() {
        let ti = TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Datetimen, 8, None));
        let buf = encode_with_ti(&ti, ColumnData::DateTime2(None)).unwrap();

        let reader = &mut buf.into_sql_read_bytes();
        // Just needs to decode a null cleanly.
        ColumnData::decode(reader, &ti).await.unwrap();
    }

    // ----- Numeric into a Money column (lines 840-847) -----

    #[tokio::test]
    async fn numeric_with_varlen_money() {
        let ti = TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Money, 8, None));
        // Numeric 3.5 (value 35000, scale 4) -> money.
        let buf = encode_with_ti(
            &ti,
            ColumnData::Numeric(Some(Numeric::new_with_scale(35000, 4))),
        )
        .unwrap();

        let reader = &mut buf.into_sql_read_bytes();
        let nd = ColumnData::decode(reader, &ti).await.unwrap();
        assert_eq!(
            nd,
            ColumnData::Numeric(Some(Numeric::new_with_scale(35000, 4)))
        );
    }

    #[tokio::test]
    async fn none_numeric_with_varlen_money() {
        let ti = TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Money, 8, None));
        let buf = encode_with_ti(&ti, ColumnData::Numeric(None)).unwrap();

        let reader = &mut buf.into_sql_read_bytes();
        let nd = ColumnData::decode(reader, &ti).await.unwrap();
        assert_eq!(nd, ColumnData::Numeric(None));
    }

    // ----- Numeric rescale on encode (lines 859-879) -----

    #[tokio::test]
    async fn numeric_scaled_up_to_column_scale() {
        // Column scale 2 > value scale 0: value 23 -> 2300 at scale 2.
        let ti = TypeInfo::VarLenSizedPrecision {
            ty: VarLenType::Numericn,
            size: 17,
            precision: 18,
            scale: 2,
        };
        let buf = encode_with_ti(
            &ti,
            ColumnData::Numeric(Some(Numeric::new_with_scale(23, 0))),
        )
        .unwrap();

        let reader = &mut buf.into_sql_read_bytes();
        let nd = ColumnData::decode(reader, &ti).await.unwrap();
        assert_eq!(
            nd,
            ColumnData::Numeric(Some(Numeric::new_with_scale(2300, 2)))
        );
    }

    #[tokio::test]
    async fn numeric_scaled_down_to_column_scale() {
        // Column scale 2 < value scale 4: 1.2345 -> 1.23 (rounded half away).
        let ti = TypeInfo::VarLenSizedPrecision {
            ty: VarLenType::Numericn,
            size: 17,
            precision: 18,
            scale: 2,
        };
        let buf = encode_with_ti(
            &ti,
            ColumnData::Numeric(Some(Numeric::new_with_scale(12345, 4))),
        )
        .unwrap();

        let reader = &mut buf.into_sql_read_bytes();
        let nd = ColumnData::decode(reader, &ti).await.unwrap();
        assert_eq!(
            nd,
            ColumnData::Numeric(Some(Numeric::new_with_scale(123, 2)))
        );
    }

    #[tokio::test]
    async fn numeric_the_column_cannot_hold_is_an_error() {
        let decimal = |precision, scale| TypeInfo::VarLenSizedPrecision {
            ty: VarLenType::Decimaln,
            size: 17,
            precision,
            scale,
        };
        let num = |v, s| ColumnData::Numeric(Some(Numeric::new_with_scale(v, s)));

        // Rounding overflowed i128 on the first two; 999.995 rounds to
        // 1000.00, which decimal(5,2) cannot hold, nor 12345 or 1234.56.
        for (ti, data) in [
            (decimal(38, 2), num(i128::MAX, 38)),
            (decimal(38, 2), num(i128::MIN, 38)),
            (decimal(5, 2), num(999_995, 3)),
            (decimal(5, 2), num(-999_995, 3)),
            (decimal(5, 2), num(12_345, 0)),
            (decimal(5, 2), num(123_456, 2)),
            (decimal(38, 0), num(10i128.pow(38), 0)),
        ] {
            let err = encode_with_ti(&ti, data.clone()).unwrap_err();
            assert!(matches!(err, Error::Conversion(_)), "{data:?}: {err:?}");
        }

        for (ti, data, expected) in [
            (decimal(5, 2), num(999_994, 3), (99_999, 2)),
            (decimal(5, 2), num(-999_994, 3), (-99_999, 2)),
            (decimal(5, 2), num(-999, 0), (-99_900, 2)),
            (
                decimal(38, 0),
                num(10i128.pow(38) - 1, 0),
                (10i128.pow(38) - 1, 0),
            ),
            (decimal(38, 37), num(-15, 38), (-2, 37)),
        ] {
            let buf = encode_with_ti(&ti, data).unwrap();
            let nd = ColumnData::decode(&mut buf.into_sql_read_bytes(), &ti)
                .await
                .unwrap();
            assert_eq!(nd, num(expected.0, expected.1));
        }
    }

    // ----- SQL_VARIANT encode (lines 897-898) -----

    #[tokio::test]
    async fn ssvariant_round_trip_i32() {
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(VarLenType::SSVariant, 0, None)),
            ColumnData::I32(Some(42)),
        )
        .await;
    }

    fn int_ti(len: usize) -> TypeInfo {
        TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Intn, len, None))
    }

    fn float_ti(len: usize) -> TypeInfo {
        TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Floatn, len, None))
    }

    async fn decode_with(ti: &TypeInfo, buf: BytesMut) -> ColumnData<'static> {
        ColumnData::decode(&mut buf.into_sql_read_bytes(), ti)
            .await
            .unwrap()
    }

    #[tokio::test]
    async fn bulk_integer_is_written_at_the_column_width() {
        let buf = encode_with_ti(&int_ti(8), ColumnData::I32(Some(-5))).unwrap();
        assert_eq!(
            &buf[..],
            &[8, 0xfb, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff]
        );
        assert_eq!(
            decode_with(&int_ti(8), buf).await,
            ColumnData::I64(Some(-5))
        );

        let ti = TypeInfo::FixedLen(FixedLenType::Int4);
        let buf = encode_with_ti(&ti, ColumnData::I16(Some(-1))).unwrap();
        assert_eq!(&buf[..], &[0xff; 4]);

        let buf = encode_with_ti(&int_ti(1), ColumnData::I64(Some(200))).unwrap();
        assert_eq!(
            decode_with(&int_ti(1), buf).await,
            ColumnData::U8(Some(200))
        );

        let buf = encode_with_ti(&int_ti(2), ColumnData::I32(None)).unwrap();
        assert_eq!(&buf[..], &[0]);
    }

    #[test]
    fn bulk_integer_out_of_the_column_range_errors() {
        for (len, value) in [
            (4, ColumnData::I64(Some(5_000_000_000))),
            (2, ColumnData::I32(Some(70_000))),
            (1, ColumnData::I32(Some(300))),
            (1, ColumnData::I16(Some(-1))),
        ] {
            let err = encode_with_ti(&int_ti(len), value).unwrap_err();
            assert!(matches!(err, Error::Conversion(_)), "got {err:?}");
        }
    }

    #[tokio::test]
    async fn bulk_float_is_written_at_the_column_width() {
        let buf = encode_with_ti(&float_ti(8), ColumnData::F32(Some(1.5))).unwrap();
        assert_eq!(
            decode_with(&float_ti(8), buf).await,
            ColumnData::F64(Some(1.5))
        );

        let ti = TypeInfo::FixedLen(FixedLenType::Float4);
        let buf = encode_with_ti(&ti, ColumnData::F64(Some(1.5))).unwrap();
        assert_eq!(&buf[..], &1.5f32.to_le_bytes());

        let err = encode_with_ti(&float_ti(4), ColumnData::F64(Some(1e300))).unwrap_err();
        assert!(matches!(err, Error::Conversion(_)), "got {err:?}");
    }

    #[test]
    fn a_non_finite_float_is_rejected_before_it_is_sent() {
        for value in [f64::NAN, f64::INFINITY, f64::NEG_INFINITY] {
            // As an RPC parameter, at both widths, and into a bulk column.
            for data in [
                ColumnData::F64(Some(value)),
                ColumnData::F32(Some(value as f32)),
            ] {
                let mut buf = BytesMut::new();
                let err = data
                    .clone()
                    .encode(&mut BytesMutWithTypeInfo::new(&mut buf))
                    .unwrap_err();
                assert!(matches!(err, Error::Conversion(_)), "{data:?}: {err:?}");

                let err = encode_with_ti(&float_ti(8), data.clone()).unwrap_err();
                assert!(matches!(err, Error::Conversion(_)), "{data:?}: {err:?}");
            }
        }
    }

    #[test]
    fn bulk_null_into_not_null_column_errors() {
        let ti = TypeInfo::FixedLen(FixedLenType::Int4);
        let err = encode_with_ti(&ti, ColumnData::I32(None)).unwrap_err();
        assert!(matches!(err, Error::BulkInput(_)), "got {err:?}");
    }

    #[tokio::test]
    async fn bulk_not_null_money() {
        let money = TypeInfo::FixedLen(FixedLenType::Money);
        let money4 = TypeInfo::FixedLen(FixedLenType::Money4);
        let exact = |v| ColumnData::Numeric(Some(Numeric::new_with_scale(v, 4)));

        let buf = encode_with_ti(&money, ColumnData::F64(Some(-1234.5))).unwrap();
        assert_eq!(decode_with(&money, buf).await, exact(-12_345_000));
        let buf = encode_with_ti(&money4, ColumnData::F64(Some(1.25))).unwrap();
        assert_eq!(decode_with(&money4, buf).await, exact(12_500));

        // Exact from a Numeric; through f64, .5806 would become .5807.
        let max_but_one = i64::MAX as i128 - 1;
        let buf = encode_with_ti(&money, exact(max_but_one)).unwrap();
        assert_eq!(decode_with(&money, buf).await, exact(max_but_one));

        assert!(encode_with_ti(&money4, ColumnData::F64(Some(300_000.0))).is_err());
    }

    /// `hh:mm:ss` plus 10^-7 s increments on 2024-05-06.
    #[cfg(feature = "tds73")]
    fn dt2(h: u64, m: u64, s: u64, increments: u64) -> ColumnData<'static> {
        let time = ((h * 60 + m) * 60 + s) * 10_000_000 + increments;
        ColumnData::DateTime2(Some(DateTime2::new(Date::new(739_011), Time::new(time, 7))))
    }

    #[cfg(feature = "tds73")]
    async fn as_datetime(ti: &TypeInfo, value: ColumnData<'static>) -> ColumnData<'static> {
        decode_with(ti, encode_with_ti(ti, value).unwrap()).await
    }

    #[cfg(feature = "tds73")]
    #[tokio::test]
    async fn bulk_datetime2_into_datetime_rounds_like_cast() {
        let ti = TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Datetimen, 8, None));
        let day = 739_011 - DAYS_YEAR_1_TO_1900 as i32;

        // CAST(datetime2 AS datetime) on SQL Server 2022: .1234567 -> .123,
        // .9983333 -> .997, .9983334 -> next day.
        assert_eq!(
            as_datetime(&ti, dt2(13, 14, 15, 1_234_567)).await,
            ColumnData::DateTime(Some(DateTime::new(day, 47_655 * 300 + 37)))
        );
        assert_eq!(
            as_datetime(&ti, dt2(23, 59, 59, 9_983_333)).await,
            ColumnData::DateTime(Some(DateTime::new(day, 86_400 * 300 - 1)))
        );
        assert_eq!(
            as_datetime(&ti, dt2(23, 59, 59, 9_983_334)).await,
            ColumnData::DateTime(Some(DateTime::new(day + 1, 0)))
        );
    }

    #[cfg(feature = "tds73")]
    #[tokio::test]
    async fn bulk_datetime_range_starts_in_1753() {
        let ti = TypeInfo::FixedLen(FixedLenType::Datetime);
        let at =
            |days| ColumnData::DateTime2(Some(DateTime2::new(Date::new(days), Time::new(0, 7))));

        // 1753-01-01 is day 639_905 after 0001-01-01.
        assert_eq!(
            as_datetime(&ti, at(639_905)).await,
            ColumnData::DateTime(Some(DateTime::new(-53_690, 0)))
        );
        let err = encode_with_ti(&ti, at(639_904)).unwrap_err();
        assert!(matches!(err, Error::Conversion(_)), "got {err:?}");
    }

    #[cfg(feature = "tds73")]
    #[tokio::test]
    async fn bulk_smalldatetime_rounds_like_cast() {
        let ti = TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Datetimen, 4, None));
        let day = (739_011 - DAYS_YEAR_1_TO_1900) as u16;

        // CAST(datetime2 AS smalldatetime): 29.999 s rounds down, 30 s up.
        assert_eq!(
            as_datetime(&ti, dt2(10, 0, 29, 9_990_000)).await,
            ColumnData::SmallDateTime(Some(SmallDateTime::new(day, 600)))
        );
        assert_eq!(
            as_datetime(&ti, dt2(10, 0, 30, 0)).await,
            ColumnData::SmallDateTime(Some(SmallDateTime::new(day, 601)))
        );
        assert_eq!(
            as_datetime(&ti, dt2(23, 59, 30, 0)).await,
            ColumnData::SmallDateTime(Some(SmallDateTime::new(day + 1, 0)))
        );

        let ti = TypeInfo::FixedLen(FixedLenType::Datetime4);
        let buf = encode_with_ti(&ti, ColumnData::DateTime(Some(DateTime::new(5, 9_000)))).unwrap();
        assert_eq!(
            decode_with(&ti, buf).await,
            ColumnData::SmallDateTime(Some(SmallDateTime::new(5, 1)))
        );
    }

    #[cfg(feature = "tds73")]
    #[tokio::test]
    async fn bulk_time_is_written_at_the_column_scale() {
        let ti = TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Timen, 3, None));
        let buf = encode_with_ti(&ti, ColumnData::Time(Some(Time::new(1_234_567, 7)))).unwrap();
        assert_eq!(&buf[..], &[4, 123, 0, 0, 0]);

        let ti = TypeInfo::VarLenSized(VarLenContext::new(VarLenType::DatetimeOffsetn, 0, None));
        let dto = DateTimeOffset::new(
            DateTime2::new(Date::new(739_011), Time::new(15_000_000, 7)),
            60,
        );
        let buf = encode_with_ti(&ti, ColumnData::DateTimeOffset(Some(dto))).unwrap();
        // length, 3 time bytes (1 s), 3 date bytes, 2 offset bytes.
        assert_eq!(buf[0], 8);
        assert_eq!(&buf[1..4], &[1, 0, 0]);

        assert_eq!(
            rescale_time(Time::new(5, 0), 7).unwrap().increments(),
            50_000_000
        );
        assert!(rescale_time(Time::new(5, 9), 7).is_err());
    }

    #[cfg(feature = "tds73")]
    #[test]
    fn bulk_times_from_24_hours_up_are_errors() {
        let column = |ty, len| TypeInfo::VarLenSized(VarLenContext::new(ty, len, None));
        let at = |increments| DateTime2::new(Date::new(739_011), Time::new(increments, 7));

        // u64::MAX overflowed converting to datetime's 1/300 s.
        for increments in [u64::MAX, Time::LEAP_SECOND, 864_000_000_000] {
            for (ti, data) in [
                (
                    column(VarLenType::Timen, 3),
                    ColumnData::Time(Some(Time::new(increments, 7))),
                ),
                (
                    column(VarLenType::Datetime2, 7),
                    ColumnData::DateTime2(Some(at(increments))),
                ),
                (
                    column(VarLenType::Datetimen, 8),
                    ColumnData::DateTime2(Some(at(increments))),
                ),
                (
                    column(VarLenType::Datetimen, 4),
                    ColumnData::DateTime2(Some(at(increments))),
                ),
            ] {
                let err = encode_with_ti(&ti, data).unwrap_err();
                assert!(matches!(err, Error::Conversion(_)), "{ti:?}: {err:?}");
            }
        }
    }

    #[tokio::test]
    async fn bulk_max_values_over_64k() {
        let collation = Some(Collation::new(13632521, 52));
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(
                VarLenType::BigVarChar,
                0xffff,
                collation,
            )),
            ColumnData::String(Some("a".repeat(70_000).into())),
        )
        .await;
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(VarLenType::NVarchar, 0xffff, collation)),
            ColumnData::String(Some("a".repeat(40_000).into())),
        )
        .await;
        test_round_trip(
            TypeInfo::VarLenSized(VarLenContext::new(VarLenType::BigVarBin, 0xffff, None)),
            ColumnData::Binary(Some(vec![7u8; 70_000].into())),
        )
        .await;
    }

    #[tokio::test]
    async fn bulk_image() {
        let ti = TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Image, 0x7fff_ffff, None));
        test_round_trip(ti.clone(), ColumnData::Binary(Some(vec![1u8, 2, 3].into()))).await;

        let buf = encode_with_ti(&ti, ColumnData::Binary(None)).unwrap();
        assert_eq!(&buf[..], &[0]);
    }

    #[test]
    fn bulk_varchar_uses_the_column_code_page() {
        // Greek_CI_AS: windows-1253.
        let ti = TypeInfo::VarLenSized(VarLenContext::new(
            VarLenType::BigVarChar,
            10,
            Some(Collation::new(0x00d0_0408, 0)),
        ));
        let buf = encode_with_ti(&ti, ColumnData::String(Some("\u{3b1}\u{3b2}".into()))).unwrap();
        assert_eq!(&buf[..], &[2, 0, 0xe1, 0xe2]);
    }

    /// 2^30 units, one more than the 2^31 - 1 bytes of xml and ntext hold.
    /// Their byte counts were u32s, which overflowed from 2^31 units.
    #[test]
    fn utf16_values_over_2_gib_are_errors() {
        let s = "a".repeat(1 << 30);
        let collation = Some(Collation::new(0x00d0_0409, 52));

        let ntext = TypeInfo::VarLenSized(VarLenContext::new(
            VarLenType::NText,
            0x7fff_fffe,
            collation,
        ));
        let err = encode_with_ti(&ntext, ColumnData::String(Some(s.as_str().into())))
            .map(drop)
            .unwrap_err();
        assert!(matches!(err, Error::BulkInput(_)), "{err:?}");

        let xml = TypeInfo::Xml {
            schema: None,
            size: 0xfffffffffffffffe_usize,
        };
        let err = encode_with_ti(&xml, ColumnData::String(Some(s.as_str().into())))
            .map(drop)
            .unwrap_err();
        assert!(matches!(err, Error::Conversion(_)), "{err:?}");

        let err = ColumnData::Xml(Some(Cow::Owned(XmlData::new(s))))
            .encode_rpc_param(&mut BytesMut::new(), false, None)
            .unwrap_err();
        assert!(matches!(err, Error::Conversion(_)), "{err:?}");
    }

    /// One byte more than varbinary(max) and image hold; `as u32` wrapped
    /// from 4 GiB. The zeroed pages are never touched.
    #[test]
    fn binaries_over_2_gib_are_errors() {
        let big = vec![0u8; 1 << 31];

        for out in [false, true] {
            let err = ColumnData::Binary(Some(big.as_slice().into()))
                .encode_rpc_param(&mut BytesMut::new(), out, None)
                .unwrap_err();
            assert!(matches!(err, Error::Conversion(_)), "{out}: {err:?}");
        }

        let image = TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Image, 0x7fff_ffff, None));
        let err = encode_with_ti(&image, ColumnData::Binary(Some(big.as_slice().into())))
            .map(drop)
            .unwrap_err();
        assert!(matches!(err, Error::BulkInput(_)), "{err:?}");
    }

    #[test]
    #[ignore = "allocates 4 GiB"]
    fn text_over_2_gib_is_an_error() {
        let utf8 = Some(Collation::new(0x0400_0409, 0));
        let text = TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Text, 0x7fff_ffff, utf8));
        let s = "a".repeat(1 << 31);
        let err = encode_with_ti(&text, ColumnData::String(Some(s.into())))
            .map(drop)
            .unwrap_err();
        assert!(matches!(err, Error::BulkInput(_)), "{err:?}");
    }

    /// Rounding carried them into the next day: 24:00:00.5 was stored as
    /// 00:00:00.5 the day after.
    #[test]
    fn bulk_datetimes_past_midnight_are_errors() {
        let datetimen =
            |len| TypeInfo::VarLenSized(VarLenContext::new(VarLenType::Datetimen, len, None));
        let past = ColumnData::DateTime(Some(DateTime::new(0, 25_920_150)));
        let leap = ColumnData::DateTime(Some(DateTime::new(0, u32::MAX)));
        let small = ColumnData::SmallDateTime(Some(SmallDateTime::new(0, 1440)));

        for (ti, data) in [
            (TypeInfo::FixedLen(FixedLenType::Datetime), past.clone()),
            (datetimen(8), leap.clone()),
            (TypeInfo::FixedLen(FixedLenType::Datetime4), past),
            (datetimen(4), leap),
            (TypeInfo::FixedLen(FixedLenType::Datetime), small.clone()),
            (datetimen(4), small),
        ] {
            let err = encode_with_ti(&ti, data).unwrap_err();
            assert!(matches!(err, Error::Conversion(_)), "{ti:?}: {err:?}");
        }
    }
}