oracledb_protocol/thin/

bind.rs

#![forbid(unsafe_code)]

use super::*;

impl BindValue {
    pub(crate) fn is_output_only(&self) -> bool {
        matches!(self, BindValue::Output { .. })
            || matches!(self, BindValue::ReturnOutput { .. })
            || matches!(self, BindValue::ObjectOutput { .. })
            || matches!(self, BindValue::Array { values, .. } if values.is_empty())
    }

    pub(crate) fn is_return_output(&self) -> bool {
        matches!(self, BindValue::ReturnOutput { .. })
            || matches!(
                self,
                BindValue::ObjectOutput {
                    is_return: true,
                    ..
                }
            )
    }
}

pub(crate) fn write_bind_metadata_with_type(
    writer: &mut TtcWriter,
    value: &BindValue,
    ora_type_num: u8,
    csfrm: u8,
    buffer_size: u32,
    ttc_field_version: u8,
) -> Result<()> {
    let (flags, max_elements) = match value {
        BindValue::Array { max_elements, .. } => {
            (TNS_BIND_USE_INDICATORS | TNS_BIND_ARRAY, *max_elements)
        }
        _ => (TNS_BIND_USE_INDICATORS, 0),
    };
    // JSON binds advertise a TNS_JSON_MAX_LENGTH prefetch buffer (reference
    // base.pyx:1398-1400) so a returned/out OSON image streams inline.
    let buffer_size = if ora_type_num == ORA_TYPE_NUM_JSON {
        TNS_JSON_MAX_LENGTH
    } else {
        buffer_size
    };
    writer.write_u8(ora_type_num);
    writer.write_u8(flags);
    writer.write_u8(0);
    writer.write_u8(0);
    writer.write_ub4(buffer_size);
    writer.write_ub4(max_elements);
    let cont_flags = if matches!(
        ora_type_num,
        ORA_TYPE_NUM_CLOB | ORA_TYPE_NUM_BLOB | ORA_TYPE_NUM_VECTOR | ORA_TYPE_NUM_JSON
    ) {
        TNS_LOB_PREFETCH_FLAG
    } else {
        0
    };
    writer.write_ub8(cont_flags);
    if let BindValue::ObjectOutput { oid, version, .. }
    | BindValue::ObjectInput { oid, version, .. } = value
    {
        writer.write_bytes_with_two_lengths(Some(oid))?;
        writer.write_ub4(*version);
    } else {
        writer.write_ub4(0);
        writer.write_ub2(0);
    }
    if csfrm != 0 {
        writer.write_ub2(TNS_CHARSET_UTF8);
    } else {
        writer.write_ub2(0);
    }
    writer.write_u8(csfrm);
    // max chars (LOB prefetch length): VECTOR advertises TNS_VECTOR_MAX_LENGTH
    // so the server prefetches the image inline (reference base.pyx)
    let lob_prefetch_length = match ora_type_num {
        ORA_TYPE_NUM_VECTOR => TNS_VECTOR_MAX_LENGTH,
        ORA_TYPE_NUM_JSON => TNS_JSON_MAX_LENGTH,
        _ => 0,
    };
    writer.write_ub4(lob_prefetch_length);
    // oaccolid is gated on the negotiated field version >= 12.2 (reference
    // messages/base.pyx:1429). A pre-12.2 server (ttc field version 7, Oracle
    // 12.1 — above our accept floor of 315) does not read it, so writing it
    // unconditionally adds a stray ub4 per bind column and corrupts the bind
    // metadata. The symmetric read path already gates the same skip (see
    // fetch.rs `>= TNS_CCAP_FIELD_VERSION_12_2`); the write side had missed it.
    // Our live matrix floor is 18c (field version 11), where the branch always
    // fired, so the miss stayed invisible.
    if version_gates::carries_oaccolid(ttc_field_version) {
        writer.write_ub4(0); // oaccolid
    }
    Ok(())
}

pub fn bind_value_type_info(value: &BindValue) -> Option<BindTypeInfo> {
    let (ora_type_num, csfrm, buffer_size) = match value {
        BindValue::Null => return None,
        // An IN OUT bind carries an input value (whose Oracle type/charset form
        // is the bind's type) but must reserve enough buffer for the value the
        // server writes back, so its effective buffer is the larger of the
        // input's natural size and the caller's requested output size. A NULL
        // input carries no type; fall back to VARCHAR so the OUT slot is still
        // typed and sized.
        BindValue::InOut {
            value,
            out_buffer_size,
        } => {
            let (ora_type_num, csfrm, input_size) = match bind_value_type_info(value) {
                Some(info) => (info.ora_type_num, info.csfrm, info.buffer_size),
                None => (ORA_TYPE_NUM_VARCHAR, CS_FORM_IMPLICIT, 0),
            };
            return Some(BindTypeInfo {
                ora_type_num,
                csfrm,
                buffer_size: input_size.max(*out_buffer_size).max(1),
            });
        }
        BindValue::TypedNull {
            ora_type_num,
            csfrm,
            buffer_size,
        }
        | BindValue::Output {
            ora_type_num,
            csfrm,
            buffer_size,
        }
        | BindValue::ReturnOutput {
            ora_type_num,
            csfrm,
            buffer_size,
        } => (*ora_type_num, *csfrm, (*buffer_size).max(1)),
        BindValue::ObjectOutput { buffer_size, .. }
        | BindValue::ObjectInput { buffer_size, .. } => {
            (ORA_TYPE_NUM_OBJECT, 0, (*buffer_size).max(1))
        }
        // values larger than 32767 bytes keep the VARCHAR/RAW bind type with
        // a large buffer size; the chunked length encoding carries the data
        // (reference always derives VARCHAR/RAW from str/bytes values —
        // metadata.pyx from_value — and never switches the bind type to LONG,
        // which the server rejects for PL/SQL LOB parameters with ORA-01460)
        BindValue::Text(value) => {
            let buffer_size = u32::try_from(value.chars().count())
                .unwrap_or(u32::MAX)
                .saturating_mul(4)
                .max(1);
            (ORA_TYPE_NUM_VARCHAR, CS_FORM_IMPLICIT, buffer_size)
        }
        BindValue::Raw(value) => {
            let buffer_size = u32::try_from(value.len()).unwrap_or(u32::MAX).max(1);
            (ORA_TYPE_NUM_RAW, 0, buffer_size)
        }
        BindValue::Lob {
            ora_type_num,
            csfrm,
            ..
        } => (*ora_type_num, *csfrm, 1),
        BindValue::Number(_) => (ORA_TYPE_NUM_NUMBER, 0, ORA_TYPE_SIZE_NUMBER),
        BindValue::BinaryInteger(_) => (ORA_TYPE_NUM_BINARY_INTEGER, 0, ORA_TYPE_SIZE_NUMBER),
        BindValue::Boolean(_) => (ORA_TYPE_NUM_BOOLEAN, 0, ORA_TYPE_SIZE_BOOLEAN),
        BindValue::BinaryDouble(_) => (ORA_TYPE_NUM_BINARY_DOUBLE, 0, ORA_TYPE_SIZE_BINARY_DOUBLE),
        BindValue::BinaryFloat(_) => (ORA_TYPE_NUM_BINARY_FLOAT, 0, ORA_TYPE_SIZE_BINARY_FLOAT),
        BindValue::IntervalDS { .. } => (ORA_TYPE_NUM_INTERVAL_DS, 0, ORA_TYPE_SIZE_INTERVAL_DS),
        BindValue::IntervalYM { .. } => (ORA_TYPE_NUM_INTERVAL_YM, 0, ORA_TYPE_SIZE_INTERVAL_YM),
        BindValue::DateTime { .. } => (ORA_TYPE_NUM_DATE, 0, ORA_TYPE_SIZE_DATE),
        BindValue::Timestamp { ora_type_num, .. } => (
            *ora_type_num,
            0,
            if *ora_type_num == ORA_TYPE_NUM_TIMESTAMP_TZ {
                ORA_TYPE_SIZE_TIMESTAMP_TZ
            } else {
                ORA_TYPE_SIZE_TIMESTAMP
            },
        ),
        BindValue::TimestampTz { .. } => (ORA_TYPE_NUM_TIMESTAMP_TZ, 0, ORA_TYPE_SIZE_TIMESTAMP_TZ),
        BindValue::Array {
            ora_type_num,
            csfrm,
            buffer_size,
            ..
        } => (*ora_type_num, *csfrm, (*buffer_size).max(1)),
        // reference base.pyx _write_column_metadata: VECTOR binds advertise a
        // TNS_VECTOR_MAX_LENGTH prefetch buffer and the LOB-prefetch cont flag
        BindValue::Vector(_) => (ORA_TYPE_NUM_VECTOR, 0, TNS_VECTOR_MAX_LENGTH),
        // JSON binds: the reference DB_TYPE_JSON var has buffer_size_factor 0, so
        // its metadata buffer_size is small and the OSON value is written inline
        // (not deferred to the "long" bind section). The TNS_JSON_MAX_LENGTH
        // prefetch buffer is applied only in the wire metadata writer, not here,
        // so the long/non-long bind-data ordering matches the reference.
        BindValue::Json(_) => (ORA_TYPE_NUM_JSON, 0, 1),
        BindValue::Cursor { .. } => (ORA_TYPE_NUM_CURSOR, 0, 4),
    };
    Some(BindTypeInfo {
        ora_type_num,
        csfrm,
        buffer_size,
    })
}

pub fn define_metadata_from_bind(source: &ColumnMetadata, value: &BindValue) -> ColumnMetadata {
    let Some(mut info) = bind_value_type_info(value) else {
        return source.clone();
    };
    if source.ora_type_num == ORA_TYPE_NUM_CLOB
        && matches!(
            info.ora_type_num,
            ORA_TYPE_NUM_CHAR | ORA_TYPE_NUM_LONG | ORA_TYPE_NUM_VARCHAR
        )
    {
        info.ora_type_num = ORA_TYPE_NUM_LONG;
        if source.csfrm != 0 {
            info.csfrm = source.csfrm;
        }
    }
    let mut metadata = source.clone();
    metadata.ora_type_num = info.ora_type_num;
    metadata.csfrm = info.csfrm;
    if info.ora_type_num == ORA_TYPE_NUM_LONG {
        metadata.buffer_size = TNS_MAX_LONG_LENGTH;
        metadata.max_size = 0;
    } else {
        metadata.buffer_size = info.buffer_size.max(1);
        metadata.max_size = info.buffer_size.max(1);
    }
    metadata
}

/// When the same query is re-executed after a column's data type changed to
/// CLOB/BLOB but the previous execution fetched the column as a char/raw
/// type, the server streams the data as LONG/LONG RAW (same as a define of
/// CLOB/BLOB as string/bytes); the fetch metadata must follow (reference
/// impl/thin/messages/base.pyx:820-845 `_adjust_metadata`). Returns `true`
/// when the metadata was adjusted.
pub fn adjust_refetch_metadata(previous: &ColumnMetadata, current: &mut ColumnMetadata) -> bool {
    if current.ora_type_num == ORA_TYPE_NUM_CLOB
        && matches!(
            previous.ora_type_num,
            ORA_TYPE_NUM_CHAR | ORA_TYPE_NUM_LONG | ORA_TYPE_NUM_VARCHAR
        )
    {
        current.ora_type_num = ORA_TYPE_NUM_LONG;
        current.csfrm = previous.csfrm;
        current.buffer_size = TNS_MAX_LONG_LENGTH;
        current.max_size = 0;
        return true;
    }
    if current.ora_type_num == ORA_TYPE_NUM_BLOB
        && matches!(
            previous.ora_type_num,
            ORA_TYPE_NUM_RAW | ORA_TYPE_NUM_LONG_RAW
        )
    {
        current.ora_type_num = ORA_TYPE_NUM_LONG_RAW;
        current.csfrm = 0;
        current.buffer_size = TNS_MAX_LONG_LENGTH;
        current.max_size = 0;
        return true;
    }
    false
}

pub fn output_bind(value: BindValue) -> BindValue {
    match value {
        BindValue::ObjectOutput {
            schema,
            type_name,
            oid,
            version,
            buffer_size,
            ..
        } => BindValue::ObjectOutput {
            schema,
            type_name,
            oid,
            version,
            buffer_size: buffer_size.max(1),
            is_return: false,
        },
        value => {
            let info = bind_value_type_info(&value).unwrap_or(BindTypeInfo {
                ora_type_num: ORA_TYPE_NUM_VARCHAR,
                csfrm: CS_FORM_IMPLICIT,
                buffer_size: 1,
            });
            BindValue::Output {
                ora_type_num: info.ora_type_num,
                csfrm: info.csfrm,
                buffer_size: info.buffer_size,
            }
        }
    }
}

pub fn returning_output_bind(value: BindValue) -> BindValue {
    match value {
        BindValue::ObjectOutput {
            schema,
            type_name,
            oid,
            version,
            buffer_size,
            ..
        } => BindValue::ObjectOutput {
            schema,
            type_name,
            oid,
            version,
            buffer_size: buffer_size.max(1),
            is_return: true,
        },
        value => {
            let info = bind_value_type_info(&value).unwrap_or(BindTypeInfo {
                ora_type_num: ORA_TYPE_NUM_VARCHAR,
                csfrm: CS_FORM_IMPLICIT,
                buffer_size: 1,
            });
            BindValue::ReturnOutput {
                ora_type_num: info.ora_type_num,
                csfrm: info.csfrm,
                buffer_size: info.buffer_size,
            }
        }
    }
}

pub fn cursor_bind_template() -> BindValue {
    BindValue::TypedNull {
        ora_type_num: ORA_TYPE_NUM_CURSOR,
        csfrm: 0,
        buffer_size: 4,
    }
}

pub fn is_cursor_bind_template(value: &BindValue) -> bool {
    matches!(
        value,
        BindValue::TypedNull {
            ora_type_num: ORA_TYPE_NUM_CURSOR,
            ..
        }
    )
}

pub fn public_dbtype_name_from_type_name(type_name: &str) -> &'static str {
    match type_name {
        "NUMBER" | "DB_TYPE_NUMBER" | "int" | "float" | "Decimal" => "DB_TYPE_NUMBER",
        "NATIVE_INT" | "DB_TYPE_BINARY_INTEGER" => "DB_TYPE_BINARY_INTEGER",
        "NATIVE_FLOAT" | "DB_TYPE_BINARY_DOUBLE" => "DB_TYPE_BINARY_DOUBLE",
        "DB_TYPE_BINARY_FLOAT" | "BINARY_FLOAT" => "DB_TYPE_BINARY_FLOAT",
        "DB_TYPE_BOOLEAN" | "BOOLEAN" | "bool" => "DB_TYPE_BOOLEAN",
        "DB_TYPE_INTERVAL_DS" | "INTERVAL DAY TO SECOND" | "timedelta" => "DB_TYPE_INTERVAL_DS",
        "DB_TYPE_INTERVAL_YM" | "INTERVAL YEAR TO MONTH" | "IntervalYM" => "DB_TYPE_INTERVAL_YM",
        "DB_TYPE_BFILE" | "BFILE" => "DB_TYPE_BFILE",
        "DB_TYPE_JSON" | "JSON" => "DB_TYPE_JSON",
        "STRING" | "DB_TYPE_VARCHAR" | "str" => "DB_TYPE_VARCHAR",
        "DB_TYPE_CHAR" => "DB_TYPE_CHAR",
        "DB_TYPE_NCHAR" => "DB_TYPE_NCHAR",
        "DB_TYPE_NVARCHAR" => "DB_TYPE_NVARCHAR",
        "DB_TYPE_CLOB" | "CLOB" => "DB_TYPE_CLOB",
        "DB_TYPE_NCLOB" | "NCLOB" => "DB_TYPE_NCLOB",
        "DB_TYPE_BLOB" | "BLOB" => "DB_TYPE_BLOB",
        "DB_TYPE_LONG" | "LONG" | "LONG_STRING" => "DB_TYPE_LONG",
        "DB_TYPE_LONG_NVARCHAR" | "LONG NVARCHAR" => "DB_TYPE_LONG_NVARCHAR",
        "DB_TYPE_LONG_RAW" | "LONG RAW" | "LONG_BINARY" => "DB_TYPE_LONG_RAW",
        "DB_TYPE_RAW" | "BINARY" | "bytes" => "DB_TYPE_RAW",
        "ROWID" | "DB_TYPE_ROWID" => "DB_TYPE_ROWID",
        "DB_TYPE_UROWID" => "DB_TYPE_UROWID",
        "DATETIME" | "DB_TYPE_DATE" | "date" | "datetime" => "DB_TYPE_DATE",
        "DB_TYPE_TIMESTAMP" | "TIMESTAMP" => "DB_TYPE_TIMESTAMP",
        "DB_TYPE_TIMESTAMP_LTZ" | "TIMESTAMP WITH LOCAL TIME ZONE" => "DB_TYPE_TIMESTAMP_LTZ",
        "DB_TYPE_TIMESTAMP_TZ" | "TIMESTAMP WITH TIME ZONE" => "DB_TYPE_TIMESTAMP_TZ",
        "DB_TYPE_CURSOR" | "CURSOR" => "DB_TYPE_CURSOR",
        "DB_TYPE_VECTOR" | "VECTOR" => "DB_TYPE_VECTOR",
        _ => "DB_TYPE_VARCHAR",
    }
}

pub fn column_metadata_is_xmltype(metadata: &ColumnMetadata) -> bool {
    metadata
        .object_schema
        .as_deref()
        .is_some_and(|schema| schema.eq_ignore_ascii_case("SYS"))
        && metadata
            .object_type_name
            .as_deref()
            .is_some_and(|name| name.eq_ignore_ascii_case("XMLTYPE"))
}

pub fn public_dbtype_name_from_column_metadata(metadata: &ColumnMetadata) -> &'static str {
    if column_metadata_is_xmltype(metadata) {
        return "DB_TYPE_XMLTYPE";
    }
    match (metadata.ora_type_num, metadata.csfrm) {
        (ORA_TYPE_NUM_LONG, CS_FORM_NCHAR) => "DB_TYPE_LONG_NVARCHAR",
        (ORA_TYPE_NUM_LONG, _) => "DB_TYPE_LONG",
        (ORA_TYPE_NUM_LONG_RAW, _) => "DB_TYPE_LONG_RAW",
        (ORA_TYPE_NUM_VARCHAR, CS_FORM_NCHAR) => "DB_TYPE_NVARCHAR",
        (ORA_TYPE_NUM_CHAR, CS_FORM_NCHAR) => "DB_TYPE_NCHAR",
        (ORA_TYPE_NUM_CHAR, _) => "DB_TYPE_CHAR",
        (ORA_TYPE_NUM_VARCHAR, _) => "DB_TYPE_VARCHAR",
        (ORA_TYPE_NUM_RAW, _) => "DB_TYPE_RAW",
        (ORA_TYPE_NUM_ROWID, _) => "DB_TYPE_ROWID",
        (ORA_TYPE_NUM_UROWID, _) => "DB_TYPE_UROWID",
        (ORA_TYPE_NUM_BINARY_DOUBLE, _) => "DB_TYPE_BINARY_DOUBLE",
        (ORA_TYPE_NUM_BINARY_FLOAT, _) => "DB_TYPE_BINARY_FLOAT",
        (ORA_TYPE_NUM_BINARY_INTEGER, _) => "DB_TYPE_BINARY_INTEGER",
        (ORA_TYPE_NUM_NUMBER, _) => "DB_TYPE_NUMBER",
        (ORA_TYPE_NUM_CURSOR, _) => "DB_TYPE_CURSOR",
        (ORA_TYPE_NUM_OBJECT, _) => "DB_TYPE_OBJECT",
        (ORA_TYPE_NUM_CLOB, CS_FORM_NCHAR) => "DB_TYPE_NCLOB",
        (ORA_TYPE_NUM_CLOB, _) => "DB_TYPE_CLOB",
        (ORA_TYPE_NUM_BLOB, _) => "DB_TYPE_BLOB",
        (ORA_TYPE_NUM_BFILE, _) => "DB_TYPE_BFILE",
        (ORA_TYPE_NUM_DATE, _) => "DB_TYPE_DATE",
        (ORA_TYPE_NUM_TIMESTAMP, _) => "DB_TYPE_TIMESTAMP",
        (ORA_TYPE_NUM_TIMESTAMP_LTZ, _) => "DB_TYPE_TIMESTAMP_LTZ",
        (ORA_TYPE_NUM_TIMESTAMP_TZ, _) => "DB_TYPE_TIMESTAMP_TZ",
        (ORA_TYPE_NUM_INTERVAL_DS, _) => "DB_TYPE_INTERVAL_DS",
        (ORA_TYPE_NUM_INTERVAL_YM, _) => "DB_TYPE_INTERVAL_YM",
        (ORA_TYPE_NUM_BOOLEAN, _) => "DB_TYPE_BOOLEAN",
        (ORA_TYPE_NUM_VECTOR, _) => "DB_TYPE_VECTOR",
        (ORA_TYPE_NUM_JSON, _) => "DB_TYPE_JSON",
        _ => "DB_TYPE_VARCHAR",
    }
}

/// Mirrors the reference `DbType.default_size` / `_buffer_size_factor` table
/// (reference impl/base/types.pyx:120-440). Returns
/// `(default_size, buffer_size_factor)` for a public database type name.
pub fn public_dbtype_size_info(dbtype_name: &str) -> (u32, u32) {
    match dbtype_name {
        "DB_TYPE_BFILE" => (0, 4000),
        "DB_TYPE_BINARY_DOUBLE" => (0, ORA_TYPE_SIZE_BINARY_DOUBLE),
        "DB_TYPE_BINARY_FLOAT" => (0, ORA_TYPE_SIZE_BINARY_FLOAT),
        "DB_TYPE_BINARY_INTEGER" | "DB_TYPE_NUMBER" => (0, ORA_TYPE_SIZE_NUMBER),
        "DB_TYPE_BLOB" | "DB_TYPE_CLOB" | "DB_TYPE_NCLOB" => (0, 112),
        "DB_TYPE_BOOLEAN" => (0, ORA_TYPE_SIZE_BOOLEAN),
        "DB_TYPE_CHAR" | "DB_TYPE_NCHAR" => (2000, 4),
        "DB_TYPE_CURSOR" => (0, 4),
        "DB_TYPE_DATE" => (0, ORA_TYPE_SIZE_DATE),
        "DB_TYPE_INTERVAL_DS" => (0, ORA_TYPE_SIZE_INTERVAL_DS),
        "DB_TYPE_INTERVAL_YM" => (0, 5),
        "DB_TYPE_LONG" | "DB_TYPE_LONG_NVARCHAR" | "DB_TYPE_LONG_RAW" => (0, TNS_MAX_LONG_LENGTH),
        "DB_TYPE_NVARCHAR" | "DB_TYPE_VARCHAR" => (4000, 4),
        "DB_TYPE_RAW" => (4000, 1),
        "DB_TYPE_ROWID" => (0, ORA_TYPE_SIZE_ROWID),
        "DB_TYPE_TIMESTAMP" | "DB_TYPE_TIMESTAMP_LTZ" => (0, ORA_TYPE_SIZE_TIMESTAMP),
        "DB_TYPE_TIMESTAMP_TZ" => (0, ORA_TYPE_SIZE_TIMESTAMP_TZ),
        "DB_TYPE_JSON" | "DB_TYPE_VECTOR" => (0, 1),
        _ => (0, 0),
    }
}

/// Mirrors the reference fetch-conversion legality matrix
/// (reference impl/base/var.pyx:113-248 `_check_fetch_conversion`). Given the
/// metadata of the column being fetched and the Oracle type requested by an
/// output type handler variable, returns the metadata that should be used for
/// the wire define. Conversions that only affect the Python materialization
/// keep the original wire metadata; LOB and JSON sources adjust the define so
/// the server sends inline data. Unsupported pairs return `None` and the
/// caller is expected to raise `DPY-4007`.
pub fn check_fetch_conversion(
    source: &ColumnMetadata,
    to_ora_type_num: u8,
    to_csfrm: u8,
) -> Option<ColumnMetadata> {
    const CHAR_TYPES: [u8; 3] = [ORA_TYPE_NUM_CHAR, ORA_TYPE_NUM_LONG, ORA_TYPE_NUM_VARCHAR];
    let from = source.ora_type_num;
    let to = to_ora_type_num;
    if from == to {
        return Some(source.clone());
    }
    let supported = match from {
        ORA_TYPE_NUM_BINARY_DOUBLE | ORA_TYPE_NUM_BINARY_FLOAT => {
            matches!(
                to,
                ORA_TYPE_NUM_BINARY_INTEGER
                    | ORA_TYPE_NUM_BINARY_DOUBLE
                    | ORA_TYPE_NUM_BINARY_FLOAT
                    | ORA_TYPE_NUM_NUMBER
            ) || CHAR_TYPES.contains(&to)
        }
        ORA_TYPE_NUM_BINARY_INTEGER => to == ORA_TYPE_NUM_NUMBER || CHAR_TYPES.contains(&to),
        ORA_TYPE_NUM_BLOB => {
            if matches!(to, ORA_TYPE_NUM_RAW | ORA_TYPE_NUM_LONG_RAW) {
                let mut metadata = source.clone();
                metadata.ora_type_num = ORA_TYPE_NUM_LONG_RAW;
                metadata.csfrm = 0;
                metadata.buffer_size = TNS_MAX_LONG_LENGTH;
                metadata.max_size = 0;
                return Some(metadata);
            }
            false
        }
        ORA_TYPE_NUM_CHAR | ORA_TYPE_NUM_LONG | ORA_TYPE_NUM_VARCHAR => {
            matches!(
                to,
                ORA_TYPE_NUM_BINARY_DOUBLE
                    | ORA_TYPE_NUM_BINARY_FLOAT
                    | ORA_TYPE_NUM_NUMBER
                    | ORA_TYPE_NUM_BINARY_INTEGER
            ) || CHAR_TYPES.contains(&to)
        }
        ORA_TYPE_NUM_CLOB => {
            if CHAR_TYPES.contains(&to) {
                let mut metadata = source.clone();
                metadata.ora_type_num = ORA_TYPE_NUM_LONG;
                metadata.buffer_size = TNS_MAX_LONG_LENGTH;
                metadata.max_size = 0;
                return Some(metadata);
            }
            false
        }
        ORA_TYPE_NUM_DATE
        | ORA_TYPE_NUM_TIMESTAMP
        | ORA_TYPE_NUM_TIMESTAMP_LTZ
        | ORA_TYPE_NUM_TIMESTAMP_TZ => {
            matches!(
                to,
                ORA_TYPE_NUM_DATE
                    | ORA_TYPE_NUM_TIMESTAMP
                    | ORA_TYPE_NUM_TIMESTAMP_LTZ
                    | ORA_TYPE_NUM_TIMESTAMP_TZ
            ) || CHAR_TYPES.contains(&to)
        }
        ORA_TYPE_NUM_INTERVAL_DS | ORA_TYPE_NUM_INTERVAL_YM | ORA_TYPE_NUM_ROWID => {
            CHAR_TYPES.contains(&to)
        }
        ORA_TYPE_NUM_NUMBER => {
            matches!(
                to,
                ORA_TYPE_NUM_BINARY_INTEGER
                    | ORA_TYPE_NUM_BINARY_DOUBLE
                    | ORA_TYPE_NUM_BINARY_FLOAT
            ) || CHAR_TYPES.contains(&to)
        }
        ORA_TYPE_NUM_JSON => {
            // Native JSON (DB_TYPE_JSON) fetched as a character type via an
            // output type handler. The reference defines the column to the
            // server as VARCHAR but decodes the returned bytes as LONG: "the
            // server won't accept LONG being defined but even so it still sends
            // back LONG data" (reference impl/base/var.pyx:208-215, where
            // `_fetch_metadata.dbtype = DB_TYPE_LONG` and `return
            // DB_TYPE_VARCHAR`).
            //
            // Our wire define writer keys the VARCHAR ora_type_num off this
            // metadata, so the server accepts the define and streams the OSON
            // image inline as text. The returned data is then decoded through
            // the same `read_bytes` path used for VARCHAR/CHAR/LONG (all three
            // share identical framing in `parse_column_value`), and the
            // non-zero LONG-sized `buffer_size` set here keeps the
            // null-by-describe shortcut from firing — exactly the effect the
            // reference obtains by decoding as LONG. The handler's outconverter
            // (e.g. `json.loads`) then materializes the Python value.
            if matches!(to, ORA_TYPE_NUM_CHAR | ORA_TYPE_NUM_VARCHAR) {
                let mut metadata = source.clone();
                metadata.ora_type_num = ORA_TYPE_NUM_VARCHAR;
                metadata.csfrm = CS_FORM_IMPLICIT;
                metadata.buffer_size = TNS_MAX_LONG_LENGTH;
                metadata.max_size = 0;
                return Some(metadata);
            }
            // JSON fetched as RAW/bytes decodes the OSON image bytes directly
            // (reference var.pyx:216-218 sets `_fetch_metadata.dbtype =
            // DB_TYPE_RAW`).
            if to == ORA_TYPE_NUM_RAW {
                let mut metadata = source.clone();
                metadata.ora_type_num = ORA_TYPE_NUM_RAW;
                metadata.csfrm = 0;
                metadata.buffer_size = TNS_MAX_LONG_LENGTH;
                metadata.max_size = 0;
                return Some(metadata);
            }
            false
        }
        ORA_TYPE_NUM_VECTOR => {
            // VECTOR fetched as a character type streams its JSON text via a
            // LONG wire define; VECTOR fetched as a CLOB streams via a CLOB
            // locator (reference var.pyx:234-243).
            if CHAR_TYPES.contains(&to) {
                let mut metadata = source.clone();
                metadata.ora_type_num = ORA_TYPE_NUM_LONG;
                metadata.csfrm = CS_FORM_IMPLICIT;
                metadata.buffer_size = TNS_MAX_LONG_LENGTH;
                metadata.max_size = 0;
                return Some(metadata);
            }
            if to == ORA_TYPE_NUM_CLOB {
                let mut metadata = source.clone();
                metadata.ora_type_num = ORA_TYPE_NUM_CLOB;
                return Some(metadata);
            }
            false
        }
        _ => false,
    };
    let _ = to_csfrm;
    if supported {
        Some(source.clone())
    } else {
        None
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum BuiltinDescriptorFamily {
    Timestamp,
    TimestampTz,
    TimestampLtz,
    IntervalDs,
    IntervalYm,
}

/// Recognizes only built-in descriptor names returned by Oracle's type catalog.
///
/// `ALL_TYPE_ATTRS` returns the short `WITH TZ` and `WITH LOCAL TZ` spellings on
/// the 23ai release lane. PL/SQL metadata can spell the base type as
/// `TIMESTAMP(6)`. Fold ASCII case and whitespace, but do not use a prefix
/// match: an arbitrary ADT called `TIMESTAMP_AUDIT` must remain an ADT.
fn builtin_descriptor_family(type_name: &str) -> Option<BuiltinDescriptorFamily> {
    if let Some(timestamp_suffix) = timestamp_descriptor_suffix(type_name) {
        let mut words = timestamp_suffix.split_ascii_whitespace();
        let second = words.next();
        return match second {
            None => Some(BuiltinDescriptorFamily::Timestamp),
            Some(with) if with.eq_ignore_ascii_case("WITH") => match words.next() {
                Some(tz) if tz.eq_ignore_ascii_case("TZ") && words.next().is_none() => {
                    Some(BuiltinDescriptorFamily::TimestampTz)
                }
                Some(time) if time.eq_ignore_ascii_case("TIME") => (words
                    .next()
                    .is_some_and(|zone| zone.eq_ignore_ascii_case("ZONE"))
                    && words.next().is_none())
                .then_some(BuiltinDescriptorFamily::TimestampTz),
                Some(local) if local.eq_ignore_ascii_case("LOCAL") => match words.next() {
                    Some(tz) if tz.eq_ignore_ascii_case("TZ") && words.next().is_none() => {
                        Some(BuiltinDescriptorFamily::TimestampLtz)
                    }
                    Some(time) if time.eq_ignore_ascii_case("TIME") => (words
                        .next()
                        .is_some_and(|zone| zone.eq_ignore_ascii_case("ZONE"))
                        && words.next().is_none())
                    .then_some(BuiltinDescriptorFamily::TimestampLtz),
                    _ => None,
                },
                _ => None,
            },
            _ => None,
        };
    }

    let mut words = type_name.split_ascii_whitespace();
    let first = words.next()?;
    if first.eq_ignore_ascii_case("INTERVAL") {
        return match (words.next(), words.next(), words.next(), words.next()) {
            (Some(day), Some(to), Some(second), None)
                if day.eq_ignore_ascii_case("DAY")
                    && to.eq_ignore_ascii_case("TO")
                    && second.eq_ignore_ascii_case("SECOND") =>
            {
                Some(BuiltinDescriptorFamily::IntervalDs)
            }
            (Some(year), Some(to), Some(month), None)
                if year.eq_ignore_ascii_case("YEAR")
                    && to.eq_ignore_ascii_case("TO")
                    && month.eq_ignore_ascii_case("MONTH") =>
            {
                Some(BuiltinDescriptorFamily::IntervalYm)
            }
            _ => None,
        };
    }

    None
}

/// Returns the words after a strict `TIMESTAMP` base type, accepting at most
/// one Oracle fractional-seconds precision suffix (`(0)` through `(9)`).
fn timestamp_descriptor_suffix(type_name: &str) -> Option<&str> {
    let type_name = type_name.trim_matches(|ch: char| ch.is_ascii_whitespace());
    let (timestamp, mut suffix) = type_name
        .get(.."TIMESTAMP".len())
        .zip(type_name.get("TIMESTAMP".len()..))?;
    if !timestamp.eq_ignore_ascii_case("TIMESTAMP") {
        return None;
    }

    let has_word_separator = suffix
        .as_bytes()
        .first()
        .is_some_and(|byte| byte.is_ascii_whitespace());
    suffix = suffix.trim_start_matches(|ch: char| ch.is_ascii_whitespace());
    if suffix.is_empty() {
        return Some("");
    }
    if !suffix.starts_with('(') {
        return has_word_separator.then_some(suffix);
    }
    suffix = &suffix[1..];
    suffix = suffix.trim_start_matches(|ch: char| ch.is_ascii_whitespace());
    let precision = *suffix.as_bytes().first()?;
    if !precision.is_ascii_digit() {
        return None;
    }
    suffix = suffix[1..].trim_start_matches(|ch: char| ch.is_ascii_whitespace());
    suffix = suffix.strip_prefix(')')?;
    if suffix.is_empty() {
        return Some("");
    }
    suffix
        .as_bytes()
        .first()
        .is_some_and(|byte| byte.is_ascii_whitespace())
        .then(|| suffix.trim_matches(|ch: char| ch.is_ascii_whitespace()))
}

pub fn public_dbtype_name_from_oracle_type_name(type_name: &str) -> &'static str {
    if let Some(family) = builtin_descriptor_family(type_name) {
        return match family {
            BuiltinDescriptorFamily::Timestamp => "DB_TYPE_TIMESTAMP",
            BuiltinDescriptorFamily::TimestampTz => "DB_TYPE_TIMESTAMP_TZ",
            BuiltinDescriptorFamily::TimestampLtz => "DB_TYPE_TIMESTAMP_LTZ",
            BuiltinDescriptorFamily::IntervalDs => "DB_TYPE_INTERVAL_DS",
            BuiltinDescriptorFamily::IntervalYm => "DB_TYPE_INTERVAL_YM",
        };
    }

    let upper = type_name.to_ascii_uppercase();
    match upper.as_str() {
        "CHAR" => "DB_TYPE_CHAR",
        "NCHAR" => "DB_TYPE_NCHAR",
        "VARCHAR2" | "VARCHAR" => "DB_TYPE_VARCHAR",
        "NVARCHAR2" | "NVARCHAR" => "DB_TYPE_NVARCHAR",
        "RAW" => "DB_TYPE_RAW",
        "DATE" => "DB_TYPE_DATE",
        "CLOB" => "DB_TYPE_CLOB",
        "NCLOB" => "DB_TYPE_NCLOB",
        "BLOB" => "DB_TYPE_BLOB",
        "XMLTYPE" => "DB_TYPE_XMLTYPE",
        "BINARY_FLOAT" => "DB_TYPE_BINARY_FLOAT",
        "BINARY_DOUBLE" => "DB_TYPE_BINARY_DOUBLE",
        "NUMBER" | "INTEGER" | "SMALLINT" | "REAL" | "DOUBLE PRECISION" | "FLOAT" => {
            "DB_TYPE_NUMBER"
        }
        // PL/SQL scalar attribute/element type names returned verbatim by the
        // type catalog. Without these arms they would fall through to the ADT
        // fallback below and be misclassified as nested objects (reference
        // impl/base/types.pyx:154-175,451-455 db_type_by_ora_name).
        "BOOLEAN" | "PL/SQL BOOLEAN" => "DB_TYPE_BOOLEAN",
        "BINARY_INTEGER" | "PLS_INTEGER" | "PL/SQL BINARY INTEGER" | "PL/SQL PLS INTEGER" => {
            "DB_TYPE_BINARY_INTEGER"
        }
        "LONG" => "DB_TYPE_LONG",
        "LONG RAW" => "DB_TYPE_LONG_RAW",
        "ROWID" => "DB_TYPE_ROWID",
        "UROWID" => "DB_TYPE_UROWID",
        "BFILE" => "DB_TYPE_BFILE",
        "JSON" => "DB_TYPE_JSON",
        "VECTOR" => "DB_TYPE_VECTOR",
        // An unknown name IS a nested object type (mirrors reference
        // _create_attr only calling get_type_for_info when type_owner is set).
        _ => "DB_TYPE_OBJECT",
    }
}

pub fn dbobject_attr_precision_scale(
    type_name: &str,
    precision: Option<i8>,
    scale: Option<i8>,
) -> (i8, i8) {
    if let Some(family) = builtin_descriptor_family(type_name) {
        return match family {
            BuiltinDescriptorFamily::Timestamp
            | BuiltinDescriptorFamily::TimestampTz
            | BuiltinDescriptorFamily::TimestampLtz => (precision.unwrap_or(0), scale.unwrap_or(6)),
            BuiltinDescriptorFamily::IntervalDs => (precision.unwrap_or(2), scale.unwrap_or(6)),
            BuiltinDescriptorFamily::IntervalYm => (precision.unwrap_or(2), scale.unwrap_or(0)),
        };
    }

    match type_name.to_ascii_uppercase().as_str() {
        "NUMBER" => (
            precision.unwrap_or(if scale == Some(0) { 38 } else { 0 }),
            scale.unwrap_or(-127),
        ),
        "INTEGER" | "SMALLINT" => (precision.unwrap_or(38), scale.unwrap_or(0)),
        "REAL" => (precision.unwrap_or(63), scale.unwrap_or(-127)),
        "DOUBLE PRECISION" | "FLOAT" => (precision.unwrap_or(126), scale.unwrap_or(-127)),
        // The other built-in dictionary types intentionally retain (0, 0):
        // character/raw/LOB, DATE, binary float/double, PL/SQL scalar, LONG,
        // ROWID, BFILE, JSON, and VECTOR attributes do not use this helper's
        // precision/scale defaults. `other_builtin_descriptor_types_keep_zero`
        // enumerates that invariant; unknown names remain true ADTs.
        _ => (0, 0),
    }
}

/// Projects descriptor precision/scale into python-oracledb's public
/// `DbObjectAttr` contract.
///
/// Oracle supplies temporal descriptor defaults (for example, `(0, 6)` for a
/// bare `TIMESTAMP`), and [`dbobject_attr_precision_scale`] preserves those raw
/// values. The Python public API exposes precision and scale only for numeric
/// attributes, so its temporal attributes must remain `(0, 0)` internally in
/// order for the wrapper to present `None` for both fields.
pub fn dbobject_attr_public_precision_scale(
    type_name: &str,
    precision: Option<i8>,
    scale: Option<i8>,
) -> (i8, i8) {
    let precision_scale = dbobject_attr_precision_scale(type_name, precision, scale);
    if builtin_descriptor_family(type_name).is_some() {
        (0, 0)
    } else {
        precision_scale
    }
}

pub fn dbobject_attr_max_size(type_name: &str, length: Option<u32>) -> u32 {
    let length = length.unwrap_or(0);
    match type_name.to_ascii_uppercase().as_str() {
        "NCHAR" | "NVARCHAR2" | "NVARCHAR" => length.saturating_mul(2),
        _ => length,
    }
}

pub fn dbobject_rowtype_attr_max_size(
    type_name: &str,
    data_length: Option<u32>,
    char_length: Option<u32>,
) -> u32 {
    match type_name.to_ascii_uppercase().as_str() {
        "CHAR" | "VARCHAR" | "VARCHAR2" | "RAW" => data_length.unwrap_or(0),
        "NCHAR" | "NVARCHAR" | "NVARCHAR2" => dbobject_attr_max_size(
            type_name,
            char_length.filter(|length| *length > 0).or(data_length),
        ),
        _ => 0,
    }
}

pub fn public_dbtype_name_from_bind(value: &BindValue) -> &'static str {
    match value {
        BindValue::TypedNull {
            ora_type_num,
            csfrm,
            ..
        }
        | BindValue::Output {
            ora_type_num,
            csfrm,
            ..
        }
        | BindValue::ReturnOutput {
            ora_type_num,
            csfrm,
            ..
        }
        | BindValue::Array {
            ora_type_num,
            csfrm,
            ..
        } => public_dbtype_name_from_type_info(*ora_type_num, *csfrm),
        BindValue::ObjectOutput { .. } | BindValue::ObjectInput { .. } => "DB_TYPE_OBJECT",
        // An IN OUT bind's public type is that of its input value.
        BindValue::InOut { value, .. } => public_dbtype_name_from_bind(value),
        BindValue::Text(_) => "DB_TYPE_VARCHAR",
        BindValue::Raw(_) => "DB_TYPE_RAW",
        BindValue::Lob {
            ora_type_num,
            csfrm,
            ..
        } => match (*ora_type_num, *csfrm) {
            (ORA_TYPE_NUM_BLOB, _) => "DB_TYPE_BLOB",
            (ORA_TYPE_NUM_CLOB, CS_FORM_NCHAR) => "DB_TYPE_NCLOB",
            (ORA_TYPE_NUM_CLOB, _) => "DB_TYPE_CLOB",
            _ => "DB_TYPE_CLOB",
        },
        BindValue::Number(_) => "DB_TYPE_NUMBER",
        BindValue::BinaryInteger(_) => "DB_TYPE_BINARY_INTEGER",
        BindValue::BinaryDouble(_) => "DB_TYPE_BINARY_DOUBLE",
        BindValue::BinaryFloat(_) => "DB_TYPE_BINARY_FLOAT",
        BindValue::Boolean(_) => "DB_TYPE_BOOLEAN",
        BindValue::IntervalDS { .. } => "DB_TYPE_INTERVAL_DS",
        BindValue::IntervalYM { .. } => "DB_TYPE_INTERVAL_YM",
        BindValue::DateTime { .. } => "DB_TYPE_DATE",
        BindValue::Timestamp { ora_type_num, .. } => match *ora_type_num {
            ORA_TYPE_NUM_TIMESTAMP_LTZ => "DB_TYPE_TIMESTAMP_LTZ",
            ORA_TYPE_NUM_TIMESTAMP_TZ => "DB_TYPE_TIMESTAMP_TZ",
            _ => "DB_TYPE_TIMESTAMP",
        },
        BindValue::TimestampTz { .. } => "DB_TYPE_TIMESTAMP_TZ",
        BindValue::Vector(_) => "DB_TYPE_VECTOR",
        BindValue::Json(_) => "DB_TYPE_JSON",
        BindValue::Cursor { .. } => "DB_TYPE_CURSOR",
        BindValue::Null => "DB_TYPE_VARCHAR",
    }
}

pub fn bind_template_from_type_name(type_name: &str, size: u32) -> BindValue {
    let text_buffer_size = if size == 0 { 4000 } else { size.max(1) };
    let nchar_buffer_size = text_buffer_size.saturating_mul(4);
    match type_name {
        "NUMBER" | "DB_TYPE_NUMBER" | "int" | "float" | "Decimal" => BindValue::TypedNull {
            ora_type_num: ORA_TYPE_NUM_NUMBER,
            csfrm: 0,
            buffer_size: ORA_TYPE_SIZE_NUMBER,
        },
        "NATIVE_INT" | "DB_TYPE_BINARY_INTEGER" => BindValue::TypedNull {
            ora_type_num: ORA_TYPE_NUM_BINARY_INTEGER,
            csfrm: 0,
            buffer_size: ORA_TYPE_SIZE_NUMBER,
        },
        "NATIVE_FLOAT" | "DB_TYPE_BINARY_DOUBLE" => BindValue::TypedNull {
            ora_type_num: ORA_TYPE_NUM_BINARY_DOUBLE,
            csfrm: 0,
            buffer_size: ORA_TYPE_SIZE_BINARY_DOUBLE,
        },
        "DB_TYPE_BINARY_FLOAT" | "BINARY_FLOAT" => BindValue::TypedNull {
            ora_type_num: ORA_TYPE_NUM_BINARY_FLOAT,
            csfrm: 0,
            buffer_size: ORA_TYPE_SIZE_BINARY_FLOAT,
        },
        "DB_TYPE_BOOLEAN" | "BOOLEAN" | "bool" => BindValue::TypedNull {
            ora_type_num: ORA_TYPE_NUM_BOOLEAN,
            csfrm: 0,
            buffer_size: ORA_TYPE_SIZE_BOOLEAN,
        },
        "DB_TYPE_INTERVAL_DS" | "INTERVAL DAY TO SECOND" | "timedelta" => BindValue::TypedNull {
            ora_type_num: ORA_TYPE_NUM_INTERVAL_DS,
            csfrm: 0,
            buffer_size: ORA_TYPE_SIZE_INTERVAL_DS,
        },
        "DB_TYPE_INTERVAL_YM" | "INTERVAL YEAR TO MONTH" | "IntervalYM" => BindValue::TypedNull {
            ora_type_num: ORA_TYPE_NUM_INTERVAL_YM,
            csfrm: 0,
            buffer_size: ORA_TYPE_SIZE_INTERVAL_YM,
        },
        "STRING" | "DB_TYPE_VARCHAR" | "DB_TYPE_CHAR" | "str" => BindValue::TypedNull {
            ora_type_num: ORA_TYPE_NUM_VARCHAR,
            csfrm: CS_FORM_IMPLICIT,
            buffer_size: text_buffer_size,
        },
        "DB_TYPE_NCHAR" | "DB_TYPE_NVARCHAR" => BindValue::TypedNull {
            ora_type_num: ORA_TYPE_NUM_VARCHAR,
            csfrm: CS_FORM_NCHAR,
            buffer_size: nchar_buffer_size,
        },
        "DB_TYPE_CLOB" | "CLOB" => BindValue::TypedNull {
            ora_type_num: ORA_TYPE_NUM_LONG,
            csfrm: CS_FORM_IMPLICIT,
            buffer_size: TNS_MAX_LONG_LENGTH,
        },
        "DB_TYPE_NCLOB" | "NCLOB" => BindValue::TypedNull {
            ora_type_num: ORA_TYPE_NUM_LONG,
            csfrm: CS_FORM_NCHAR,
            buffer_size: TNS_MAX_LONG_LENGTH,
        },
        "DB_TYPE_BLOB" | "BLOB" => BindValue::TypedNull {
            ora_type_num: ORA_TYPE_NUM_LONG_RAW,
            csfrm: 0,
            buffer_size: TNS_MAX_LONG_LENGTH,
        },
        "DB_TYPE_LONG" | "LONG" | "LONG_STRING" => BindValue::TypedNull {
            ora_type_num: ORA_TYPE_NUM_LONG,
            csfrm: CS_FORM_IMPLICIT,
            buffer_size: TNS_MAX_LONG_LENGTH,
        },
        "DB_TYPE_LONG_NVARCHAR" | "LONG NVARCHAR" => BindValue::TypedNull {
            ora_type_num: ORA_TYPE_NUM_LONG,
            csfrm: CS_FORM_NCHAR,
            buffer_size: TNS_MAX_LONG_LENGTH,
        },
        "DB_TYPE_LONG_RAW" | "LONG RAW" | "LONG_BINARY" => BindValue::TypedNull {
            ora_type_num: ORA_TYPE_NUM_LONG_RAW,
            csfrm: 0,
            buffer_size: TNS_MAX_LONG_LENGTH,
        },
        "DB_TYPE_RAW" | "BINARY" | "bytes" => BindValue::TypedNull {
            ora_type_num: ORA_TYPE_NUM_RAW,
            csfrm: 0,
            buffer_size: size.max(1).max(4000),
        },
        "ROWID" | "DB_TYPE_ROWID" | "DB_TYPE_UROWID" => BindValue::TypedNull {
            ora_type_num: ORA_TYPE_NUM_VARCHAR,
            csfrm: CS_FORM_IMPLICIT,
            buffer_size: 5267,
        },
        "DATETIME" | "DB_TYPE_DATE" | "date" | "datetime" => BindValue::TypedNull {
            ora_type_num: ORA_TYPE_NUM_DATE,
            csfrm: 0,
            buffer_size: ORA_TYPE_SIZE_DATE,
        },
        "DB_TYPE_TIMESTAMP" | "TIMESTAMP" => BindValue::TypedNull {
            ora_type_num: ORA_TYPE_NUM_TIMESTAMP,
            csfrm: 0,
            buffer_size: ORA_TYPE_SIZE_TIMESTAMP,
        },
        "DB_TYPE_TIMESTAMP_LTZ" | "TIMESTAMP WITH LOCAL TIME ZONE" => BindValue::TypedNull {
            ora_type_num: ORA_TYPE_NUM_TIMESTAMP_LTZ,
            csfrm: 0,
            buffer_size: ORA_TYPE_SIZE_TIMESTAMP,
        },
        "DB_TYPE_TIMESTAMP_TZ" | "TIMESTAMP WITH TIME ZONE" => BindValue::TypedNull {
            ora_type_num: ORA_TYPE_NUM_TIMESTAMP_TZ,
            csfrm: 0,
            buffer_size: ORA_TYPE_SIZE_TIMESTAMP_TZ,
        },
        "DB_TYPE_CURSOR" | "CURSOR" => cursor_bind_template(),
        "DB_TYPE_VECTOR" | "VECTOR" => BindValue::TypedNull {
            ora_type_num: ORA_TYPE_NUM_VECTOR,
            csfrm: 0,
            buffer_size: TNS_VECTOR_MAX_LENGTH,
        },
        "DB_TYPE_JSON" | "JSON" => BindValue::TypedNull {
            ora_type_num: ORA_TYPE_NUM_JSON,
            csfrm: 0,
            buffer_size: TNS_VECTOR_MAX_LENGTH,
        },
        _ => BindValue::Null,
    }
}

pub fn dbobject_element_bind_type_info(dbtype_name: &str, max_size: u32) -> BindTypeInfo {
    let buffer_size = max_size.max(1);
    let (ora_type_num, csfrm, buffer_size) = match dbtype_name {
        "DB_TYPE_NUMBER" => (ORA_TYPE_NUM_NUMBER, 0, ORA_TYPE_SIZE_NUMBER),
        "DB_TYPE_RAW" | "DB_TYPE_BLOB" => (ORA_TYPE_NUM_RAW, 0, buffer_size.max(4000)),
        "DB_TYPE_NCHAR" | "DB_TYPE_NVARCHAR" | "DB_TYPE_NCLOB" => {
            (ORA_TYPE_NUM_VARCHAR, CS_FORM_NCHAR, buffer_size.max(4000))
        }
        "DB_TYPE_DATE" => (ORA_TYPE_NUM_DATE, 0, ORA_TYPE_SIZE_DATE),
        "DB_TYPE_TIMESTAMP" => (ORA_TYPE_NUM_TIMESTAMP, 0, ORA_TYPE_SIZE_TIMESTAMP),
        "DB_TYPE_TIMESTAMP_LTZ" => (ORA_TYPE_NUM_TIMESTAMP_LTZ, 0, ORA_TYPE_SIZE_TIMESTAMP),
        "DB_TYPE_TIMESTAMP_TZ" => (ORA_TYPE_NUM_TIMESTAMP_TZ, 0, ORA_TYPE_SIZE_TIMESTAMP_TZ),
        _ => (
            ORA_TYPE_NUM_VARCHAR,
            CS_FORM_IMPLICIT,
            buffer_size.max(4000),
        ),
    };
    BindTypeInfo {
        ora_type_num,
        csfrm,
        buffer_size,
    }
}

pub(crate) fn public_dbtype_name_from_type_info(ora_type_num: u8, csfrm: u8) -> &'static str {
    match (ora_type_num, csfrm) {
        (ORA_TYPE_NUM_BINARY_DOUBLE, _) => "DB_TYPE_BINARY_DOUBLE",
        (ORA_TYPE_NUM_BINARY_FLOAT, _) => "DB_TYPE_BINARY_FLOAT",
        (ORA_TYPE_NUM_INTERVAL_DS, _) => "DB_TYPE_INTERVAL_DS",
        (ORA_TYPE_NUM_INTERVAL_YM, _) => "DB_TYPE_INTERVAL_YM",
        (ORA_TYPE_NUM_BOOLEAN, _) => "DB_TYPE_BOOLEAN",
        (ORA_TYPE_NUM_BINARY_INTEGER, _) => "DB_TYPE_BINARY_INTEGER",
        (ORA_TYPE_NUM_NUMBER, _) => "DB_TYPE_NUMBER",
        (ORA_TYPE_NUM_CHAR, CS_FORM_NCHAR) | (ORA_TYPE_NUM_VARCHAR, CS_FORM_NCHAR) => {
            "DB_TYPE_NVARCHAR"
        }
        (ORA_TYPE_NUM_CHAR, _) => "DB_TYPE_CHAR",
        (ORA_TYPE_NUM_VARCHAR, _) => "DB_TYPE_VARCHAR",
        (ORA_TYPE_NUM_LONG, CS_FORM_NCHAR) => "DB_TYPE_LONG_NVARCHAR",
        (ORA_TYPE_NUM_LONG, _) => "DB_TYPE_LONG",
        (ORA_TYPE_NUM_LONG_RAW, _) => "DB_TYPE_LONG_RAW",
        (ORA_TYPE_NUM_RAW, _) => "DB_TYPE_RAW",
        (ORA_TYPE_NUM_DATE, _) => "DB_TYPE_DATE",
        (ORA_TYPE_NUM_TIMESTAMP, _) => "DB_TYPE_TIMESTAMP",
        (ORA_TYPE_NUM_TIMESTAMP_LTZ, _) => "DB_TYPE_TIMESTAMP_LTZ",
        (ORA_TYPE_NUM_TIMESTAMP_TZ, _) => "DB_TYPE_TIMESTAMP_TZ",
        (ORA_TYPE_NUM_CURSOR, _) => "DB_TYPE_CURSOR",
        (ORA_TYPE_NUM_OBJECT, _) => "DB_TYPE_OBJECT",
        (ORA_TYPE_NUM_VECTOR, _) => "DB_TYPE_VECTOR",
        (ORA_TYPE_NUM_JSON, _) => "DB_TYPE_JSON",
        _ => "DB_TYPE_VARCHAR",
    }
}

pub(crate) fn bind_metadata(value: &BindValue) -> (u8, u8, u32) {
    bind_value_type_info(value)
        .map(|info| (info.ora_type_num, info.csfrm, info.buffer_size))
        .unwrap_or((ORA_TYPE_NUM_VARCHAR, CS_FORM_IMPLICIT, 1))
}

pub(crate) fn write_bind_value(writer: &mut TtcWriter, value: &BindValue, csfrm: u8) -> Result<()> {
    match value {
        BindValue::TypedNull {
            ora_type_num: ORA_TYPE_NUM_CURSOR,
            ..
        } => {
            writer.write_u8(1);
            writer.write_u8(0);
            Ok(())
        }
        // A NULL BOOLEAN bind is encoded as the two raw bytes
        // [TNS_ESCAPE_CHAR, 1], not the usual single 0 null indicator; sending
        // a plain 0 makes the server reject a PL/SQL BOOLEAN parameter with
        // PLS-00306 (reference messages/base.pyx _write_bind_params_column).
        BindValue::TypedNull {
            ora_type_num: ORA_TYPE_NUM_BOOLEAN,
            ..
        } => {
            writer.write_u8(TNS_ESCAPE_CHAR);
            writer.write_u8(1);
            Ok(())
        }
        BindValue::Null | BindValue::TypedNull { .. } => {
            writer.write_u8(0);
            Ok(())
        }
        BindValue::Output { .. } | BindValue::ReturnOutput { .. } => {
            writer.write_u8(0);
            Ok(())
        }
        // An IN OUT bind sends its input value bytes (never a null indicator);
        // the read-back is driven by the server's IO-vector direction, not by
        // anything written here. Recurse with the resolved csfrm (which the
        // metadata derived from this same inner value).
        BindValue::InOut { value, .. } => write_bind_value(writer, value, csfrm),
        BindValue::ObjectOutput { .. } => {
            // NULL object image (empty OUT bind): reference messages/base.pyx
            // 1462-1468.
            writer.write_ub4(0);
            writer.write_ub4(0);
            writer.write_ub4(0);
            writer.write_ub2(0);
            writer.write_ub4(0);
            writer.write_ub4(TNS_OBJ_TOP_LEVEL);
            Ok(())
        }
        BindValue::ObjectInput { oid, image, .. } => write_dbobject_bind(writer, oid, image),
        BindValue::Text(value) => {
            // The common implicit/single-byte-charset case writes the &str bytes
            // straight through (no throwaway Vec); only NCHAR re-encodes to UTF-16
            // and needs the owned buffer. Byte-identical to encode_text_value,
            // which for the non-NCHAR path is exactly `value.as_bytes().to_vec()`.
            if csfrm == CS_FORM_NCHAR {
                let bytes = encode_text_value(value, csfrm);
                writer.write_bytes_with_length(&bytes)
            } else {
                writer.write_bytes_with_length(value.as_bytes())
            }
        }
        BindValue::Raw(value) => writer.write_bytes_with_length(value),
        BindValue::Lob { locator, .. } => writer.write_bytes_with_two_lengths(Some(locator)),
        BindValue::Number(value) | BindValue::BinaryInteger(value) => {
            let bytes = encode_number_text(value)?;
            writer.write_bytes_with_length(&bytes)
        }
        // reference encode_boolean (impl/base/encoders.pyx:99-111): true is
        // the two bytes [1, 1]; false is the single byte [0]
        BindValue::Boolean(value) => {
            let bytes: &[u8] = if *value { &[1, 1] } else { &[0] };
            writer.write_bytes_with_length(bytes)
        }
        BindValue::BinaryDouble(value) => {
            let bytes = encode_binary_double(*value);
            writer.write_bytes_with_length(&bytes)
        }
        BindValue::BinaryFloat(value) => {
            let bytes = encode_binary_float(*value as f32);
            writer.write_bytes_with_length(&bytes)
        }
        BindValue::IntervalDS {
            days,
            seconds,
            microseconds,
        } => {
            let nanoseconds = microseconds
                .checked_mul(1000)
                .ok_or(ProtocolError::TtcDecode(
                    "INTERVAL DS fractional seconds out of range",
                ))?;
            let bytes = encode_interval_ds(*days, *seconds, nanoseconds)?;
            writer.write_bytes_with_length(&bytes)
        }
        BindValue::IntervalYM { years, months } => {
            let bytes = encode_interval_ym(*years, *months)?;
            writer.write_bytes_with_length(&bytes)
        }
        BindValue::DateTime {
            year,
            month,
            day,
            hour,
            minute,
            second,
        } => {
            let bytes = encode_oracle_date(*year, *month, *day, *hour, *minute, *second)?;
            writer.write_bytes_with_length(&bytes)
        }
        BindValue::Timestamp {
            year,
            month,
            day,
            hour,
            minute,
            second,
            nanosecond,
            ora_type_num,
        } => {
            let bytes = if matches!(*ora_type_num, ORA_TYPE_NUM_TIMESTAMP_TZ) {
                encode_oracle_timestamp_tz(
                    *year,
                    *month,
                    *day,
                    *hour,
                    *minute,
                    *second,
                    *nanosecond,
                )?
            } else {
                encode_oracle_timestamp(*year, *month, *day, *hour, *minute, *second, *nanosecond)?
            };
            writer.write_bytes_with_length(&bytes)
        }
        BindValue::TimestampTz {
            year,
            month,
            day,
            hour,
            minute,
            second,
            nanosecond,
            offset_minutes,
        } => {
            let bytes = encode_oracle_timestamp_tz_with_offset(
                *year,
                *month,
                *day,
                *hour,
                *minute,
                *second,
                *nanosecond,
                *offset_minutes,
            )?;
            writer.write_bytes_with_length(&bytes)
        }
        BindValue::Array {
            values,
            csfrm: array_csfrm,
            ..
        } => {
            writer.write_ub4(u32::try_from(values.len()).map_err(|_| {
                ProtocolError::InvalidPacketLength {
                    length: values.len(),
                    minimum: 0,
                }
            })?);
            for value in values {
                match value {
                    Some(value) => write_bind_value(writer, value, *array_csfrm)?,
                    None => writer.write_u8(0),
                }
            }
            Ok(())
        }
        // reference WriteBuffer.write_vector: a QLocator carrying the image
        // length, then the image bytes-with-length
        BindValue::Vector(vector) => {
            let image = crate::vector::encode_vector_checked(vector)?;
            crate::vector::write_vector_image(writer, &image)
        }
        // reference WriteBuffer.write_oson: a QLocator carrying the OSON image
        // length, then the image bytes-with-length (same framing as VECTOR).
        BindValue::Json(image) => crate::vector::write_vector_image(writer, image),
        BindValue::Cursor { cursor_id } => {
            if *cursor_id == 0 {
                writer.write_u8(1);
                writer.write_u8(0);
            } else {
                writer.write_ub4(1);
                writer.write_ub4(*cursor_id);
            }
            Ok(())
        }
    }
}

pub(crate) fn encode_text_value(value: &str, csfrm: u8) -> Vec<u8> {
    if csfrm == CS_FORM_NCHAR {
        let mut bytes = Vec::with_capacity(value.len().saturating_mul(2));
        for unit in value.encode_utf16() {
            bytes.extend_from_slice(&unit.to_be_bytes());
        }
        bytes
    } else {
        value.as_bytes().to_vec()
    }
}

#[cfg(test)]
mod bind_tests {
    use super::*;

    const ORACLE_23AI_DESCRIPTOR_FIXTURE: &str =
        include_str!("../../tests/golden/oracle_23ai_all_type_attrs_tstz.txt");

    fn written_bytes(value: &BindValue, csfrm: u8) -> Vec<u8> {
        let mut writer = TtcWriter::new();
        write_bind_value(&mut writer, value, csfrm).expect("write bind value");
        writer.into_bytes()
    }

    #[test]
    fn captured_23ai_descriptor_fixture_normalizes_precision_scale() {
        let mut captured_rows = 0;
        for line in ORACLE_23AI_DESCRIPTOR_FIXTURE
            .lines()
            .filter(|line| line.starts_with("ATTR|"))
        {
            captured_rows += 1;
            let fields: Vec<_> = line.split('|').collect();
            assert_eq!(fields.len(), 8, "fixture row shape: {line}");
            assert_eq!(fields[1], "SYS", "fixture owner: {line}");
            assert_eq!(fields[2], "C23G_TSTZ_TYPE", "fixture type: {line}");

            let precision = (fields[6] != "NULL")
                .then(|| fields[6].parse::<i8>().expect("fixture precision is an i8"));
            let scale = (fields[7] != "NULL")
                .then(|| fields[7].parse::<i8>().expect("fixture scale is an i8"));
            let (expected_dbtype, expected_precision_scale) = match fields[4] {
                "TS_DEFAULT" => ("DB_TYPE_TIMESTAMP", (0, 6)),
                "TS_3" => ("DB_TYPE_TIMESTAMP", (0, 3)),
                "TSTZ_DEFAULT" => ("DB_TYPE_TIMESTAMP_TZ", (0, 6)),
                "TSTZ_3" => ("DB_TYPE_TIMESTAMP_TZ", (0, 3)),
                "TSLTZ_DEFAULT" => ("DB_TYPE_TIMESTAMP_LTZ", (0, 6)),
                "TSLTZ_3" => ("DB_TYPE_TIMESTAMP_LTZ", (0, 3)),
                "IDS_DEFAULT" => ("DB_TYPE_INTERVAL_DS", (2, 6)),
                "IDS_9_3" => ("DB_TYPE_INTERVAL_DS", (9, 3)),
                "IYM_DEFAULT" => ("DB_TYPE_INTERVAL_YM", (2, 0)),
                "IYM_9" => ("DB_TYPE_INTERVAL_YM", (9, 0)),
                name => panic!("unexpected captured attribute {name}"),
            };

            assert_eq!(
                public_dbtype_name_from_oracle_type_name(fields[5]),
                expected_dbtype,
                "captured raw type name: {line}"
            );
            assert_eq!(
                dbobject_attr_precision_scale(fields[5], precision, scale),
                expected_precision_scale,
                "captured raw precision/scale: {line}"
            );
        }
        assert_eq!(captured_rows, 10, "captured descriptor row count");
    }

    #[test]
    fn descriptor_normalizer_folds_only_builtin_grammar() {
        assert_eq!(
            builtin_descriptor_family(" \ttImEsTaMp  WITH\nLOCAL\tTZ  "),
            Some(BuiltinDescriptorFamily::TimestampLtz)
        );
        assert_eq!(
            builtin_descriptor_family(" timestamp ( 6 ) with time zone "),
            Some(BuiltinDescriptorFamily::TimestampTz)
        );
        assert_eq!(
            public_dbtype_name_from_oracle_type_name("TIMESTAMP(6)"),
            "DB_TYPE_TIMESTAMP"
        );
        assert_eq!(
            builtin_descriptor_family("INTERVAL  YEAR\tTO MONTH"),
            Some(BuiltinDescriptorFamily::IntervalYm)
        );
        for name in [
            "TIMESTAMP_AUDIT",
            "TIMESTAMP(10)",
            "TIMESTAMP(6)WITH TZ",
            "TIMESTAMP WITH TZ EXTRA",
            "INTERVAL DAY TO SECOND EXTRA",
        ] {
            assert_eq!(builtin_descriptor_family(name), None, "{name}");
            assert_eq!(
                public_dbtype_name_from_oracle_type_name(name),
                "DB_TYPE_OBJECT",
                "{name} remains an ADT"
            );
            assert_eq!(dbobject_attr_precision_scale(name, None, None), (0, 0));
        }
    }

    #[test]
    fn public_metadata_hides_temporal_descriptor_defaults() {
        for name in [
            "TIMESTAMP",
            "TIMESTAMP(6)",
            "TIMESTAMP WITH TZ",
            "TIMESTAMP WITH LOCAL TZ",
            "INTERVAL DAY TO SECOND",
            "INTERVAL YEAR TO MONTH",
        ] {
            assert_ne!(
                dbobject_attr_precision_scale(name, None, None),
                (0, 0),
                "raw descriptor metadata for {name}"
            );
            assert_eq!(
                dbobject_attr_public_precision_scale(name, None, None),
                (0, 0),
                "DbObjectAttr exposes None precision/scale for {name}"
            );
        }
    }

    #[test]
    fn number_descriptor_precision_scale_is_unchanged() {
        assert_eq!(
            dbobject_attr_precision_scale("NUMBER", None, None),
            (0, -127),
            "unconstrained NUMBER keeps its existing defaults"
        );
        assert_eq!(
            dbobject_attr_precision_scale("NUMBER", None, Some(0)),
            (38, 0),
            "scale-zero NUMBER keeps its existing precision default"
        );
        assert_eq!(
            dbobject_attr_precision_scale("NUMBER", Some(9), Some(3)),
            (9, 3),
            "explicit NUMBER metadata is preserved"
        );
    }

    #[test]
    fn other_builtin_descriptor_types_keep_zero_precision_scale() {
        for name in [
            "CHAR",
            "NCHAR",
            "VARCHAR2",
            "VARCHAR",
            "NVARCHAR2",
            "NVARCHAR",
            "RAW",
            "DATE",
            "CLOB",
            "NCLOB",
            "BLOB",
            "XMLTYPE",
            "BINARY_FLOAT",
            "BINARY_DOUBLE",
            "BOOLEAN",
            "PL/SQL BOOLEAN",
            "BINARY_INTEGER",
            "PLS_INTEGER",
            "PL/SQL BINARY INTEGER",
            "PL/SQL PLS INTEGER",
            "LONG",
            "LONG RAW",
            "ROWID",
            "UROWID",
            "BFILE",
            "JSON",
            "VECTOR",
        ] {
            assert_eq!(
                dbobject_attr_precision_scale(name, Some(9), Some(3)),
                (0, 0),
                "{name} intentionally has no precision/scale default"
            );
        }
    }

    // An IN OUT NUMBER bind sends its input value (not a null indicator) and its
    // metadata reserves an output slot. NUMBER is a fixed 22-byte type, so the
    // input and output sizes coincide; the point here is that the value bytes
    // are on the wire exactly as a plain IN bind writes them.
    #[test]
    fn in_out_number_writes_input_and_reserves_output_slot() {
        let inout = BindValue::InOut {
            value: Box::new(BindValue::Number("21".to_string())),
            out_buffer_size: ORA_TYPE_SIZE_NUMBER,
        };
        // Not an output-only placeholder: the input value is sent, never a null.
        assert!(!inout.is_output_only(), "IN OUT carries an input value");
        assert!(!inout.is_return_output(), "IN OUT is not a function RETURN");

        let info = bind_value_type_info(&inout).expect("IN OUT NUMBER has type info");
        assert_eq!(info.ora_type_num, ORA_TYPE_NUM_NUMBER);
        assert_eq!(info.csfrm, 0);
        assert_eq!(info.buffer_size, ORA_TYPE_SIZE_NUMBER);

        // The bytes on the wire are byte-identical to a plain IN NUMBER(21):
        // an IN OUT bind is encoded as an input bind.
        assert_eq!(
            written_bytes(&inout, 0),
            written_bytes(&BindValue::Number("21".to_string()), 0),
            "IN OUT sends the input value, same bytes as a plain IN bind"
        );
    }

    // An IN OUT VARCHAR must size its bind buffer for the *returned* value, which
    // is typically larger than the (possibly short) input. The metadata takes the
    // max of the input's natural size and the requested output size; the value
    // bytes are still just the input text.
    #[test]
    fn in_out_varchar_sizes_output_slot_and_writes_input_text() {
        // input "ab" natural size = 2 chars * 4 = 8; the OUT slot wants 200.
        let inout = BindValue::InOut {
            value: Box::new(BindValue::Text("ab".to_string())),
            out_buffer_size: 200,
        };
        let info = bind_value_type_info(&inout).expect("IN OUT VARCHAR has type info");
        assert_eq!(info.ora_type_num, ORA_TYPE_NUM_VARCHAR);
        assert_eq!(info.csfrm, CS_FORM_IMPLICIT);
        assert_eq!(
            info.buffer_size, 200,
            "the OUT slot is sized for the returned value, not the short input"
        );

        // The written bytes are exactly the input text (same as a plain IN Text).
        assert_eq!(
            written_bytes(&inout, CS_FORM_IMPLICIT),
            written_bytes(&BindValue::Text("ab".to_string()), CS_FORM_IMPLICIT),
        );

        // When the input is the larger side, its natural size wins.
        let big_input = BindValue::InOut {
            value: Box::new(BindValue::Text("abcdefghij".to_string())), // 10 * 4 = 40
            out_buffer_size: 4,
        };
        assert_eq!(
            bind_value_type_info(&big_input).unwrap().buffer_size,
            40,
            "buffer is max(input natural size, requested output size)"
        );
    }

    // The read-back metadata a server response is decoded against
    // (`bind_column_metadata`) must carry the IN OUT bind's Oracle type and the
    // reserved output buffer, so the returned value parses correctly.
    #[test]
    fn in_out_read_back_metadata_matches_output_slot() {
        let inout = BindValue::InOut {
            value: Box::new(BindValue::Text("ab".to_string())),
            out_buffer_size: 200,
        };
        let column = crate::thin::bind_column_metadata(&inout);
        assert_eq!(column.ora_type_num, ORA_TYPE_NUM_VARCHAR);
        assert_eq!(column.csfrm, CS_FORM_IMPLICIT);
        assert_eq!(column.buffer_size, 200);
        assert_eq!(column.max_size, 200);
        assert!(!column.is_array);
    }
}