//! DuckDB-over-Quack protocol/client implementation.
//!
//! This module contains the client-side protocol model and binary DuckDB message
//! encoding/decoding used by the Quack catalog backend. Catalog semantics stay in
//! `quack.rs`; this module is transport/protocol plumbing only.

use std::fmt;

use crate::{Result, StorageError, ducklake_sql_string_literal, format_uuid_bytes};

pub const DEFAULT_PORT: u16 = 9494;
pub const DEFAULT_SSL: bool = true;
pub const QUACK_VERSION: u64 = 1;
pub(crate) const DUCKDB_MESSAGE_TERMINATOR_FIELD_ID: u16 = 0xffff;
pub(crate) const DUCKDB_INVALID_OPTIONAL_IDX: u64 = u64::MAX;

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct QuackUri {
    raw: String,
    host: String,
    port: u16,
    ipv6: bool,
    ssl: bool,
    http_url: String,
}

impl QuackUri {
    pub fn parse(raw: impl Into<String>) -> Result<Self> {
        Self::parse_with_ssl(raw, DEFAULT_SSL)
    }

    pub fn parse_for_client(raw: impl Into<String>) -> Result<Self> {
        let raw = raw.into();
        let initial = Self::parse_with_ssl(raw.clone(), DEFAULT_SSL)?;
        Self::parse_with_ssl(raw, !initial.is_local())
    }

    pub fn parse_with_ssl(raw: impl Into<String>, ssl: bool) -> Result<Self> {
        let raw = raw.into().trim().to_string();
        let mut remainder = if let Some(remainder) = raw.strip_prefix("quack://") {
            remainder.to_string()
        } else if let Some(remainder) = raw.strip_prefix("quack:") {
            remainder.to_string()
        } else {
            return Err(StorageError::Backend(
                "Invalid DuckDB Quack RPC URI, needs to start with 'quack:'".into(),
            ));
        };

        if remainder.is_empty() {
            return Err(StorageError::Backend("Missing hostname".into()));
        }

        let mut host = String::new();
        let mut ipv6 = false;
        let mut port = DEFAULT_PORT;

        if let Some(after_open) = remainder.strip_prefix('[') {
            let Some(close_idx) = after_open.find(']') else {
                return Err(StorageError::Backend(
                    "Invalid IPv6 URL, missing ']'".into(),
                ));
            };
            ipv6 = true;
            host = after_open[..close_idx].to_string();
            if host.is_empty() {
                return Err(StorageError::Backend("Missing IPv6 Address".into()));
            }
            remainder = after_open[close_idx + 1..].to_string();
        }

        if let Some(colon_idx) = remainder.find(':') {
            let port_str = &remainder[colon_idx + 1..];
            if port_str.is_empty() {
                return Err(StorageError::Backend("Invalid Port".into()));
            }
            let raw_port = port_str
                .parse::<u16>()
                .map_err(|_| StorageError::Backend("Invalid Port".into()))?;
            if raw_port == 0 {
                return Err(StorageError::Backend("Invalid Port".into()));
            }
            port = raw_port;
            remainder.truncate(colon_idx);
        }

        if !ipv6 {
            host = remainder;
            if host.is_empty() {
                return Err(StorageError::Backend("Missing hostname".into()));
            }
        }

        let scheme = if ssl { "https" } else { "http" };
        let http_host = if ipv6 {
            format!("[{host}]")
        } else {
            host.clone()
        };

        Ok(Self {
            raw,
            host,
            port,
            ipv6,
            ssl,
            http_url: format!("{scheme}://{http_host}:{port}"),
        })
    }

    pub fn raw(&self) -> &str {
        &self.raw
    }

    pub fn host(&self) -> &str {
        &self.host
    }

    pub fn port(&self) -> u16 {
        self.port
    }

    pub fn is_ipv6(&self) -> bool {
        self.ipv6
    }

    pub fn ssl(&self) -> bool {
        self.ssl
    }

    pub fn http_url(&self) -> &str {
        &self.http_url
    }

    pub fn canonical_uri(&self) -> String {
        format!("quack:{}:{}", self.host, self.port)
    }

    pub fn is_local(&self) -> bool {
        self.host.eq_ignore_ascii_case("localhost")
            || self.host == "127.0.0.1"
            || self.host == "::1"
    }

    pub fn quack_endpoint(&self) -> String {
        format!("{}/quack", self.http_url)
    }
}

#[repr(u8)]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum MessageType {
    Invalid = 0,
    ConnectionRequest = 1,
    ConnectionResponse = 2,
    PrepareRequest = 3,
    PrepareResponse = 4,
    FetchRequest = 7,
    FetchResponse = 8,
    AppendRequest = 9,
    SuccessResponse = 10,
    DisconnectMessage = 11,
    ErrorResponse = 100,
}

impl TryFrom<u8> for MessageType {
    type Error = StorageError;

    fn try_from(value: u8) -> Result<Self> {
        let message_type = match value {
            0 => Self::Invalid,
            1 => Self::ConnectionRequest,
            2 => Self::ConnectionResponse,
            3 => Self::PrepareRequest,
            4 => Self::PrepareResponse,
            7 => Self::FetchRequest,
            8 => Self::FetchResponse,
            9 => Self::AppendRequest,
            10 => Self::SuccessResponse,
            11 => Self::DisconnectMessage,
            100 => Self::ErrorResponse,
            other => {
                return Err(StorageError::Backend(format!(
                    "unsupported Quack message type {other}"
                )));
            }
        };
        Ok(message_type)
    }
}

impl MessageType {
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Invalid => "INVALID",
            Self::ConnectionRequest => "CONNECTION_REQUEST",
            Self::ConnectionResponse => "CONNECTION_RESPONSE",
            Self::PrepareRequest => "PREPARE_REQUEST",
            Self::PrepareResponse => "PREPARE_RESPONSE",
            Self::FetchRequest => "FETCH_REQUEST",
            Self::FetchResponse => "FETCH_RESPONSE",
            Self::AppendRequest => "APPEND_REQUEST",
            Self::SuccessResponse => "SUCCESS_RESPONSE",
            Self::DisconnectMessage => "DISCONNECT_MESSAGE",
            Self::ErrorResponse => "ERROR_RESPONSE",
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MessageHeader {
    pub message_type: MessageType,
    pub connection_id: String,
    pub client_query_id: Option<u64>,
}

impl MessageHeader {
    pub fn new(message_type: MessageType, connection_id: impl Into<String>) -> Self {
        Self {
            message_type,
            connection_id: connection_id.into(),
            client_query_id: None,
        }
    }

    pub fn with_client_query_id(mut self, client_query_id: u64) -> Self {
        self.client_query_id = Some(client_query_id);
        self
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DuckDbPropertyKind {
    MessageType,
    String,
    OptionalIdx,
    Idx,
    HugeInt,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct DuckDbPropertyDescriptor {
    pub id: u32,
    pub name: &'static str,
    pub kind: DuckDbPropertyKind,
    pub with_default: bool,
}

impl DuckDbPropertyDescriptor {
    pub(crate) const fn new(
        id: u32,
        name: &'static str,
        kind: DuckDbPropertyKind,
        with_default: bool,
    ) -> Self {
        Self {
            id,
            name,
            kind,
            with_default,
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum DuckDbPropertyValue {
    MessageType(MessageType),
    String(String),
    OptionalIdx(Option<u64>),
    Idx(u64),
    HugeInt(i128),
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DuckDbProperty {
    pub descriptor: DuckDbPropertyDescriptor,
    pub value: DuckDbPropertyValue,
}

impl DuckDbProperty {
    pub(crate) fn new(descriptor: DuckDbPropertyDescriptor, value: DuckDbPropertyValue) -> Self {
        Self { descriptor, value }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DuckDbMessagePlan {
    pub header: Vec<DuckDbProperty>,
    pub body: Vec<DuckDbProperty>,
}

#[derive(Default)]
pub(crate) struct DuckDbBinaryWriter {
    pub(crate) output: Vec<u8>,
}

impl DuckDbBinaryWriter {
    pub(crate) fn finish(self) -> Vec<u8> {
        self.output
    }

    pub(crate) fn write_object(&mut self, properties: &[DuckDbProperty]) -> Result<()> {
        for property in properties {
            self.write_property(property)?;
        }
        self.write_field_id(DUCKDB_MESSAGE_TERMINATOR_FIELD_ID);
        Ok(())
    }

    pub(crate) fn write_property(&mut self, property: &DuckDbProperty) -> Result<()> {
        if property.descriptor.with_default && Self::is_default(&property.value) {
            return Ok(());
        }
        let field_id = u16::try_from(property.descriptor.id)
            .map_err(|_| StorageError::Backend("DuckDB serializer field id exceeds u16".into()))?;
        self.write_field_id(field_id);
        self.write_value(&property.value);
        Ok(())
    }

    pub(crate) fn is_default(value: &DuckDbPropertyValue) -> bool {
        match value {
            DuckDbPropertyValue::MessageType(message_type) => *message_type == MessageType::Invalid,
            DuckDbPropertyValue::String(value) => value.is_empty(),
            DuckDbPropertyValue::OptionalIdx(value) => value.is_none(),
            DuckDbPropertyValue::Idx(value) => *value == 0,
            DuckDbPropertyValue::HugeInt(value) => *value == 0,
        }
    }

    pub(crate) fn write_field_id(&mut self, field_id: u16) {
        self.output.extend_from_slice(&field_id.to_le_bytes());
    }

    pub(crate) fn write_value(&mut self, value: &DuckDbPropertyValue) {
        match value {
            DuckDbPropertyValue::MessageType(value) => {
                self.write_unsigned_leb128(u64::from(*value as u8));
            }
            DuckDbPropertyValue::String(value) => self.write_string(value),
            DuckDbPropertyValue::OptionalIdx(value) => {
                self.write_unsigned_leb128(value.unwrap_or(DUCKDB_INVALID_OPTIONAL_IDX));
            }
            DuckDbPropertyValue::Idx(value) => self.write_unsigned_leb128(*value),
            DuckDbPropertyValue::HugeInt(value) => self.write_hugeint(*value),
        }
    }

    pub(crate) fn write_string(&mut self, value: &str) {
        self.write_unsigned_leb128(value.len() as u64);
        self.output.extend_from_slice(value.as_bytes());
    }

    pub(crate) fn write_hugeint(&mut self, value: i128) {
        self.write_signed_leb128((value >> 64) as i64);
        self.write_unsigned_leb128(value as u64);
    }

    pub(crate) fn write_unsigned_leb128(&mut self, mut value: u64) {
        loop {
            let mut byte = (value & 0x7f) as u8;
            value >>= 7;
            if value != 0 {
                byte |= 0x80;
            }
            self.output.push(byte);
            if value == 0 {
                break;
            }
        }
    }

    pub(crate) fn write_signed_leb128(&mut self, mut value: i64) {
        loop {
            let byte = (value as u8) & 0x7f;
            value >>= 7;
            let sign_bit_set = (byte & 0x40) != 0;
            if (value == 0 && !sign_bit_set) || (value == -1 && sign_bit_set) {
                self.output.push(byte);
                break;
            }
            self.output.push(byte | 0x80);
        }
    }
}

pub(crate) struct DuckDbBinaryReader<'a> {
    input: &'a [u8],
    position: usize,
    buffered_field: Option<u16>,
}

impl<'a> DuckDbBinaryReader<'a> {
    pub(crate) fn new(input: &'a [u8]) -> Self {
        Self {
            input,
            position: 0,
            buffered_field: None,
        }
    }

    pub(crate) fn finish(self) -> Result<()> {
        if self.position == self.input.len() && self.buffered_field.is_none() {
            return Ok(());
        }
        Err(StorageError::Backend(
            "trailing bytes in Quack binary message".into(),
        ))
    }

    pub(crate) fn read_header(&mut self) -> Result<MessageHeader> {
        let message_type = self.read_message_type_property(1, "type")?;
        let connection_id = self.read_optional_string_property(2, "connection_id")?;
        let client_query_id = self.read_optional_idx_property(3, "client_query_id")?;
        self.read_object_end()?;
        Ok(MessageHeader {
            message_type,
            connection_id,
            client_query_id,
        })
    }

    pub(crate) fn read_response(&mut self, header: MessageHeader) -> Result<QuackResponse> {
        let response = match header.message_type {
            MessageType::ConnectionResponse => {
                let server_duckdb_version =
                    self.read_optional_string_property(1, "server_duckdb_version")?;
                let server_platform = self.read_optional_string_property(2, "server_platform")?;
                let quack_version = self.read_optional_idx_property_value(3, "quack_version")?;
                QuackResponse::Connection(ConnectionResponse {
                    header,
                    server_duckdb_version,
                    server_platform,
                    quack_version,
                })
            }
            MessageType::PrepareResponse => {
                let result_types =
                    self.read_optional_logical_type_vector_property(1, "result_types")?;
                let result_names = self.read_optional_string_vector_property(2, "result_names")?;
                let needs_more_fetch = self.read_optional_bool_property(3, "needs_more_fetch")?;
                let results = self.read_optional_data_chunk_vector_property(4, "results")?;
                let result_uuid = self.read_hugeint_property(5, "result_uuid")?;
                QuackResponse::Prepare(PrepareResponse {
                    header,
                    result_types,
                    result_names,
                    needs_more_fetch,
                    results,
                    result_uuid,
                })
            }
            MessageType::FetchResponse => {
                let results = self.read_optional_data_chunk_vector_property(1, "results")?;
                let batch_index = self.read_optional_idx_property(2, "batch_index")?;
                QuackResponse::Fetch(FetchResponse {
                    header,
                    results,
                    batch_index,
                })
            }
            MessageType::SuccessResponse => QuackResponse::Success(SuccessResponse { header }),
            MessageType::ErrorResponse => {
                let message = self.read_optional_string_property(1, "message")?;
                QuackResponse::Error(ErrorResponse { header, message })
            }
            other => {
                return Err(StorageError::Unsupported(format!(
                    "Quack response decoding for {} is not supported",
                    other.as_str()
                )));
            }
        };
        self.read_object_end()?;
        Ok(response)
    }

    pub(crate) fn read_optional_string_property(
        &mut self,
        field_id: u16,
        tag: &str,
    ) -> Result<String> {
        if !self.consume_optional_field(field_id)? {
            return Ok(String::new());
        }
        self.read_string().map_err(|err| {
            StorageError::Backend(format!(
                "failed to read DuckDB string property {tag}:{field_id}: {err}"
            ))
        })
    }

    pub(crate) fn read_optional_string_vector_property(
        &mut self,
        field_id: u16,
        _tag: &str,
    ) -> Result<Vec<String>> {
        if !self.consume_optional_field(field_id)? {
            return Ok(Vec::new());
        }
        let count = self.read_unsigned_leb128()? as usize;
        let mut values = Vec::with_capacity(count);
        for _ in 0..count {
            values.push(self.read_string()?);
        }
        Ok(values)
    }

    pub(crate) fn read_optional_logical_type_vector_property(
        &mut self,
        field_id: u16,
        _tag: &str,
    ) -> Result<Vec<String>> {
        if !self.consume_optional_field(field_id)? {
            return Ok(Vec::new());
        }
        let count = self.read_unsigned_leb128()?;
        let count = usize::try_from(count)
            .map_err(|_| StorageError::Backend("DuckDB list length exceeds usize".into()))?;
        let mut values = Vec::with_capacity(count);
        for _ in 0..count {
            values.push(self.read_logical_type()?.type_name());
        }
        Ok(values)
    }

    pub(crate) fn read_logical_type(&mut self) -> Result<DuckDbLogicalType> {
        self.read_property_begin(100, "id")?;
        let type_id = self.read_unsigned_leb128()?;
        let mut logical_type = duckdb_logical_type(type_id)?;
        if self.consume_optional_field(101)? {
            logical_type.extra_info = self.read_nullable_extra_type_info()?;
        }
        self.read_object_end()?;
        Ok(logical_type)
    }

    pub(crate) fn read_nullable_extra_type_info(&mut self) -> Result<Option<DuckDbExtraTypeInfo>> {
        let present = self.read_u8()? != 0;
        if !present {
            return Ok(None);
        }
        let extra_info = self.read_extra_type_info()?;
        self.read_object_end()?;
        Ok(extra_info)
    }

    pub(crate) fn read_extra_type_info(&mut self) -> Result<Option<DuckDbExtraTypeInfo>> {
        self.read_property_begin(100, "type")?;
        let info_type = self.read_unsigned_leb128()?;
        let alias = self.read_optional_string_property(101, "alias")?;
        self.read_deleted_extra_type_modifiers()?;
        self.read_extension_type_info()?;

        match info_type {
            0 => Ok(None),
            1 => Ok(Some(DuckDbExtraTypeInfo::Generic { _alias: alias })),
            2 => {
                let width = self.read_optional_u8_property(200, "width")?;
                let scale = self.read_optional_u8_property(201, "scale")?;
                Ok(Some(DuckDbExtraTypeInfo::Decimal {
                    width,
                    scale,
                    _alias: alias,
                }))
            }
            3 => {
                let collation = self.read_optional_string_property(200, "collation")?;
                Ok(Some(DuckDbExtraTypeInfo::String {
                    collation,
                    _alias: alias,
                }))
            }
            4 => {
                self.read_property_begin(200, "child_type")?;
                let child_type = self.read_logical_type()?;
                Ok(Some(DuckDbExtraTypeInfo::List {
                    child_type: Box::new(child_type),
                    _alias: alias,
                }))
            }
            5 => {
                let children = self.read_optional_child_type_list_property(200, "child_types")?;
                Ok(Some(DuckDbExtraTypeInfo::Struct {
                    children,
                    _alias: alias,
                }))
            }
            6 => {
                self.read_property_begin(200, "values_count")?;
                let values_count = self.read_usize()?;
                self.read_property_begin(201, "values")?;
                let values = self.read_string_list()?;
                if values.len() != values_count {
                    return Err(StorageError::Backend(format!(
                        "DuckDB enum value count mismatch: expected {values_count}, got {}",
                        values.len()
                    )));
                }
                Ok(Some(DuckDbExtraTypeInfo::Enum {
                    values,
                    _alias: alias,
                }))
            }
            7 => self.read_unbound_type_info(alias).map(Some),
            8 => {
                let function_name = self.read_optional_string_property(200, "function_name")?;
                self.read_property_begin(201, "return_type")?;
                let return_type = self.read_logical_type()?;
                let bound_argument_types =
                    self.read_optional_logical_type_list_property(202, "bound_argument_types")?;
                Ok(Some(DuckDbExtraTypeInfo::AggregateState {
                    function_name,
                    return_type: Box::new(return_type),
                    bound_argument_types,
                    _alias: alias,
                }))
            }
            9 => {
                self.read_property_begin(200, "child_type")?;
                let child_type = self.read_logical_type()?;
                let size = self.read_optional_u32_property(201, "size")?;
                Ok(Some(DuckDbExtraTypeInfo::Array {
                    child_type: Box::new(child_type),
                    size,
                    _alias: alias,
                }))
            }
            10 => {
                self.read_property_begin(200, "target_type")?;
                let target_type = self.read_logical_type()?;
                let cast_score = self.read_optional_idx_property_value(201, "cast_score")?;
                Ok(Some(DuckDbExtraTypeInfo::Any {
                    _target_type: Box::new(target_type),
                    _cast_score: cast_score,
                    _alias: alias,
                }))
            }
            11 => {
                self.read_property_begin(200, "constant_value")?;
                let constant_value = self.read_duckdb_value()?;
                Ok(Some(DuckDbExtraTypeInfo::IntegerLiteral {
                    constant_value,
                    _alias: alias,
                }))
            }
            12 => {
                let name = self.read_optional_string_property(200, "name")?;
                Ok(Some(DuckDbExtraTypeInfo::Template {
                    name,
                    _alias: alias,
                }))
            }
            13 => {
                self.read_property_begin(200, "crs")?;
                let crs_definition = self.read_optional_string_property(100, "definition")?;
                self.read_object_end()?;
                Ok(Some(DuckDbExtraTypeInfo::Geo {
                    crs_definition,
                    _alias: alias,
                }))
            }
            other => Err(StorageError::Unsupported(format!(
                "Quack DuckDB ExtraTypeInfoType {other} response decoding is not supported"
            ))),
        }
    }

    pub(crate) fn read_unbound_type_info(&mut self, alias: String) -> Result<DuckDbExtraTypeInfo> {
        if let Some(type_name) = self.read_optional_type_expression_property(204, "expr")? {
            return Ok(DuckDbExtraTypeInfo::Unbound {
                type_name,
                _alias: alias,
            });
        }

        let name = self.read_optional_string_property(200, "name")?;
        let catalog = self.read_optional_string_property(201, "catalog")?;
        let schema = self.read_optional_string_property(202, "schema")?;
        let modifiers =
            self.read_optional_duckdb_value_list_property(203, "user_type_modifiers")?;
        Ok(DuckDbExtraTypeInfo::Unbound {
            type_name: duckdb_unbound_type_name(&catalog, &schema, &name, &modifiers),
            _alias: alias,
        })
    }

    pub(crate) fn read_optional_type_expression_property(
        &mut self,
        field_id: u16,
        _tag: &str,
    ) -> Result<Option<String>> {
        if !self.consume_optional_field(field_id)? {
            return Ok(None);
        }
        let present = self.read_u8()? != 0;
        if !present {
            return Ok(None);
        }
        self.read_parsed_type_expression().map(Some)
    }

    pub(crate) fn read_parsed_type_expression(&mut self) -> Result<String> {
        let expression = self.read_parsed_expression()?;
        Ok(duckdb_parsed_expression_sql(&expression))
    }

    pub(crate) fn read_parsed_expression(&mut self) -> Result<DuckDbParsedExpression> {
        let header = self.read_parsed_expression_base()?;
        self.read_parsed_expression_body(header, "UnboundTypeInfo")
    }

    pub(crate) fn read_parsed_expression_body(
        &mut self,
        header: DuckDbParsedExpressionHeader,
        context: &str,
    ) -> Result<DuckDbParsedExpression> {
        let expression = match (header.expression_class, header.expression_type) {
            (2, 150) => {
                let case_checks =
                    self.read_optional_case_check_list_property(200, "case_checks")?;
                let else_expr =
                    self.read_optional_parsed_expression_box_property(201, "else_expr")?;
                DuckDbParsedExpression::Case {
                    alias: header.alias,
                    case_checks,
                    else_expr,
                }
            }
            (3, 225) => {
                let child = self.read_optional_parsed_expression_box_property(200, "child")?;
                self.read_property_begin(201, "cast_type")?;
                let cast_type = self.read_logical_type()?;
                let try_cast = self.read_optional_bool_property(202, "try_cast")?;
                DuckDbParsedExpression::Cast {
                    alias: header.alias,
                    child,
                    cast_type,
                    try_cast,
                }
            }
            (4, 203) => {
                let column_names =
                    self.read_optional_string_vector_property(200, "column_names")?;
                DuckDbParsedExpression::ColumnRef {
                    alias: header.alias,
                    column_names,
                }
            }
            (5, 25..=40) => {
                let left = self.read_optional_parsed_expression_box_property(200, "left")?;
                let right = self.read_optional_parsed_expression_box_property(201, "right")?;
                DuckDbParsedExpression::Comparison {
                    alias: header.alias,
                    expression_type: header.expression_type,
                    left,
                    right,
                }
            }
            (6, 50 | 51) => {
                let children =
                    self.read_optional_parsed_expression_list_property(200, "children")?;
                DuckDbParsedExpression::Conjunction {
                    alias: header.alias,
                    expression_type: header.expression_type,
                    children,
                }
            }
            (7, 75) => {
                self.read_property_begin(200, "value")?;
                DuckDbParsedExpression::Constant {
                    alias: header.alias,
                    value: self.read_duckdb_value()?,
                }
            }
            (8, 82) => DuckDbParsedExpression::Default {
                alias: header.alias,
            },
            (9, 140) => {
                let function_name = self.read_optional_string_property(200, "function_name")?;
                let schema = self.read_optional_string_property(201, "schema")?;
                let children =
                    self.read_optional_parsed_expression_list_property(202, "children")?;
                let filter = self.read_optional_parsed_expression_box_property(203, "filter")?;
                if self.consume_optional_field(204)? {
                    return Err(StorageError::Unsupported(format!(
                        "Quack DuckDB {context} FunctionExpression order_bys response decoding is not supported"
                    )));
                }
                let distinct = self.read_optional_bool_property(205, "distinct")?;
                let is_operator = self.read_optional_bool_property(206, "is_operator")?;
                let export_state = self.read_optional_bool_property(207, "export_state")?;
                let catalog = self.read_optional_string_property(208, "catalog")?;
                DuckDbParsedExpression::Function {
                    alias: header.alias,
                    catalog,
                    schema,
                    function_name,
                    children,
                    filter,
                    distinct,
                    is_operator,
                    export_state,
                }
            }
            (10, 12..=16 | 151..=158) => {
                let children =
                    self.read_optional_parsed_expression_list_property(200, "children")?;
                DuckDbParsedExpression::Operator {
                    alias: header.alias,
                    expression_type: header.expression_type,
                    children,
                }
            }
            (15, 76) => {
                let identifier = self.read_optional_string_property(200, "identifier")?;
                DuckDbParsedExpression::Parameter {
                    alias: header.alias,
                    identifier,
                }
            }
            (16, 230) => {
                let child = self.read_optional_parsed_expression_box_property(200, "child")?;
                let collation = self.read_optional_string_property(201, "collation")?;
                DuckDbParsedExpression::Collate {
                    alias: header.alias,
                    child,
                    collation,
                }
            }
            (18, 232) => {
                let index = self.read_optional_idx_property_value(200, "index")?;
                DuckDbParsedExpression::PositionalReference {
                    alias: header.alias,
                    index,
                }
            }
            (19, 38 | 39) => {
                let input = self.read_optional_parsed_expression_box_property(200, "input")?;
                let lower = self.read_optional_parsed_expression_box_property(201, "lower")?;
                let upper = self.read_optional_parsed_expression_box_property(202, "upper")?;
                DuckDbParsedExpression::Between {
                    alias: header.alias,
                    negated: header.expression_type == 39,
                    input,
                    lower,
                    upper,
                }
            }
            (20, 206) => {
                let lambda_idx = self.read_optional_idx_property_value(200, "lambda_idx")?;
                let column_name = self.read_optional_string_property(201, "column_name")?;
                DuckDbParsedExpression::LambdaRef {
                    alias: header.alias,
                    lambda_idx,
                    column_name,
                }
            }
            (21, 207) => {
                let catalog = self.read_optional_string_property(200, "catalog")?;
                let schema = self.read_optional_string_property(201, "schema")?;
                let type_name = self.read_optional_string_property(202, "type_name")?;
                let children =
                    self.read_optional_parsed_expression_list_property(203, "children")?;
                DuckDbParsedExpression::Type(DuckDbParsedTypeExpression {
                    alias: header.alias,
                    catalog,
                    schema,
                    type_name,
                    children,
                })
            }
            (expression_class, expression_type) => {
                return Err(StorageError::Unsupported(format!(
                    "Quack DuckDB {context} ParsedExpression class {expression_class} type {expression_type} response decoding is not supported"
                )));
            }
        };
        self.read_object_end()?;
        Ok(expression)
    }

    pub(crate) fn read_parsed_expression_base(&mut self) -> Result<DuckDbParsedExpressionHeader> {
        self.read_property_begin(100, "class")?;
        let expression_class = self.read_unsigned_leb128()?;
        self.read_property_begin(101, "type")?;
        let expression_type = self.read_unsigned_leb128()?;
        let alias = self.read_optional_string_property(102, "alias")?;
        if self.consume_optional_field(103)? {
            let _ = self.read_unsigned_leb128()?;
        }
        Ok(DuckDbParsedExpressionHeader {
            expression_class,
            expression_type,
            alias,
        })
    }

    pub(crate) fn read_optional_parsed_expression_box_property(
        &mut self,
        field_id: u16,
        _tag: &str,
    ) -> Result<Option<Box<DuckDbParsedExpression>>> {
        if !self.consume_optional_field(field_id)? {
            return Ok(None);
        }
        let present = self.read_u8()? != 0;
        if !present {
            return Ok(None);
        }
        Ok(Some(Box::new(self.read_parsed_expression()?)))
    }

    pub(crate) fn read_optional_parsed_expression_list_property(
        &mut self,
        field_id: u16,
        _tag: &str,
    ) -> Result<Vec<DuckDbParsedExpression>> {
        if !self.consume_optional_field(field_id)? {
            return Ok(Vec::new());
        }
        let count = self.read_usize()?;
        let mut values = Vec::with_capacity(count);
        for _ in 0..count {
            let present = self.read_u8()? != 0;
            if !present {
                values.push(DuckDbParsedExpression::Constant {
                    alias: String::new(),
                    value: DuckDbValue::Null,
                });
                continue;
            }
            values.push(self.read_parsed_expression()?);
        }
        Ok(values)
    }

    pub(crate) fn read_optional_case_check_list_property(
        &mut self,
        field_id: u16,
        _tag: &str,
    ) -> Result<Vec<DuckDbParsedCaseCheck>> {
        if !self.consume_optional_field(field_id)? {
            return Ok(Vec::new());
        }
        let count = self.read_usize()?;
        let mut values = Vec::with_capacity(count);
        for _ in 0..count {
            let when_expr = self.read_optional_parsed_expression_box_property(100, "when_expr")?;
            let then_expr = self.read_optional_parsed_expression_box_property(101, "then_expr")?;
            self.read_object_end()?;
            values.push(DuckDbParsedCaseCheck {
                when_expr,
                then_expr,
            });
        }
        Ok(values)
    }

    pub(crate) fn read_optional_duckdb_value_list_property(
        &mut self,
        field_id: u16,
        _tag: &str,
    ) -> Result<Vec<DuckDbValue>> {
        if !self.consume_optional_field(field_id)? {
            return Ok(Vec::new());
        }
        let count = self.read_usize()?;
        let mut values = Vec::with_capacity(count);
        for _ in 0..count {
            values.push(self.read_duckdb_value()?);
        }
        Ok(values)
    }

    pub(crate) fn read_duckdb_value(&mut self) -> Result<DuckDbValue> {
        self.read_duckdb_value_with_expected_type(None)
    }

    pub(crate) fn read_duckdb_value_with_expected_type(
        &mut self,
        expected_type: Option<&DuckDbLogicalType>,
    ) -> Result<DuckDbValue> {
        let logical_type = if self.consume_optional_field(100)? {
            self.read_logical_type()?
        } else if let Some(expected_type) = expected_type {
            expected_type.clone()
        } else {
            return Err(StorageError::Backend(
                "DuckDB Value is missing logical type".into(),
            ));
        };

        self.read_property_begin(101, "is_null")?;
        if self.read_u8()? != 0 {
            self.read_object_end()?;
            return Ok(DuckDbValue::Null);
        }

        self.read_property_begin(102, "value")?;
        let value =
            match logical_type.id {
                6 => DuckDbValue::Type(self.read_logical_type()?.type_name()),
                10 => DuckDbValue::Boolean(self.read_u8()? != 0),
                11 => {
                    DuckDbValue::Int8(i8::try_from(self.read_signed_leb128()?).map_err(|_| {
                        StorageError::Backend("DuckDB Value TINYINT exceeds i8".into())
                    })?)
                }
                12 => {
                    DuckDbValue::Int16(i16::try_from(self.read_signed_leb128()?).map_err(|_| {
                        StorageError::Backend("DuckDB Value SMALLINT exceeds i16".into())
                    })?)
                }
                13 => {
                    DuckDbValue::Int32(i32::try_from(self.read_signed_leb128()?).map_err(|_| {
                        StorageError::Backend("DuckDB Value INTEGER exceeds i32".into())
                    })?)
                }
                14 => DuckDbValue::Int64(self.read_signed_leb128()?),
                15 => {
                    DuckDbValue::DateDays(i32::try_from(self.read_signed_leb128()?).map_err(
                        |_| StorageError::Backend("DuckDB Value DATE exceeds i32".into()),
                    )?)
                }
                16 => DuckDbValue::TimeMicros(self.read_signed_leb128()?),
                17 => DuckDbValue::TimestampSeconds(self.read_signed_leb128()?),
                18 => DuckDbValue::TimestampMillis(self.read_signed_leb128()?),
                19 => DuckDbValue::TimestampMicros(self.read_signed_leb128()?),
                20 => DuckDbValue::TimestampNanos(self.read_signed_leb128()?),
                21 => self.read_duckdb_decimal_value(&logical_type)?,
                22 => {
                    let bytes: [u8; 4] = self.read_exact(4)?.try_into().map_err(|_| {
                        StorageError::Backend("truncated Quack DuckDB FLOAT value".into())
                    })?;
                    DuckDbValue::Float32(f32::from_le_bytes(bytes))
                }
                23 => {
                    let bytes: [u8; 8] = self.read_exact(8)?.try_into().map_err(|_| {
                        StorageError::Backend("truncated Quack DuckDB DOUBLE value".into())
                    })?;
                    DuckDbValue::Float64(f64::from_le_bytes(bytes))
                }
                25 => DuckDbValue::Utf8(self.read_string()?),
                26 => DuckDbValue::Blob(self.read_string()?.into_bytes()),
                27 => self.read_duckdb_interval_value()?,
                100 | 107 => {
                    let children = logical_type.struct_children()?.to_vec();
                    let values = self.read_duckdb_value_children(&children)?;
                    DuckDbValue::Struct(
                        children
                            .into_iter()
                            .zip(values)
                            .map(|(child, value)| (child.name, value))
                            .collect(),
                    )
                }
                101 => {
                    let child_type = logical_type.list_child_type()?.clone();
                    DuckDbValue::List(self.read_duckdb_repeated_value_children(&child_type, None)?)
                }
                102 => {
                    let child_type = logical_type.list_child_type()?.clone();
                    let values = self.read_duckdb_repeated_value_children(&child_type, None)?;
                    DuckDbValue::Map(duckdb_map_entries_from_structs(&values)?)
                }
                108 => {
                    let (child_type, size) = logical_type.array_child_type_and_size()?;
                    let child_type = child_type.clone();
                    DuckDbValue::Array(
                        self.read_duckdb_repeated_value_children(&child_type, Some(size))?,
                    )
                }
                28 => DuckDbValue::UInt8(u8::try_from(self.read_unsigned_leb128()?).map_err(
                    |_| StorageError::Backend("DuckDB Value UTINYINT exceeds u8".into()),
                )?),
                29 => DuckDbValue::UInt16(u16::try_from(self.read_unsigned_leb128()?).map_err(
                    |_| StorageError::Backend("DuckDB Value USMALLINT exceeds u16".into()),
                )?),
                30 => DuckDbValue::UInt32(u32::try_from(self.read_unsigned_leb128()?).map_err(
                    |_| StorageError::Backend("DuckDB Value UINTEGER exceeds u32".into()),
                )?),
                31 => DuckDbValue::UInt64(self.read_unsigned_leb128()?),
                32 => DuckDbValue::TimestampTzMicros(self.read_signed_leb128()?),
                34 => DuckDbValue::TimeTzMicros(self.read_signed_leb128()?),
                35 => DuckDbValue::TimeNanos(self.read_signed_leb128()?),
                49 => {
                    let upper = self.read_unsigned_leb128()?;
                    let lower = self.read_unsigned_leb128()?;
                    DuckDbValue::UInt128((u128::from(upper) << 64) | u128::from(lower))
                }
                50 => {
                    let upper = self.read_signed_leb128()?;
                    let lower = self.read_unsigned_leb128()?;
                    DuckDbValue::Int128(((upper as i128) << 64) | i128::from(lower))
                }
                54 => DuckDbValue::Uuid(read_duckdb_uuid_bytes_from_parts(
                    self.read_signed_leb128()?,
                    self.read_unsigned_leb128()?,
                )),
                other => {
                    return Err(StorageError::Unsupported(format!(
                        "Quack DuckDB Value decoding for LogicalTypeId {other} is not supported"
                    )));
                }
            };
        self.read_object_end()?;
        Ok(value)
    }

    pub(crate) fn read_duckdb_decimal_value(
        &mut self,
        logical_type: &DuckDbLogicalType,
    ) -> Result<DuckDbValue> {
        let (width, scale) = logical_type.decimal_width_scale().ok_or_else(|| {
            StorageError::Backend("DuckDB DECIMAL logical type is missing width/scale".into())
        })?;
        let value = match duckdb_physical_type(logical_type)? {
            DuckDbPhysicalType::Int16 => {
                i128::from(i16::try_from(self.read_signed_leb128()?).map_err(|_| {
                    StorageError::Backend("DuckDB Value DECIMAL(INT16) exceeds i16".into())
                })?)
            }
            DuckDbPhysicalType::Int32 => {
                i128::from(i32::try_from(self.read_signed_leb128()?).map_err(|_| {
                    StorageError::Backend("DuckDB Value DECIMAL(INT32) exceeds i32".into())
                })?)
            }
            DuckDbPhysicalType::Int64 => i128::from(self.read_signed_leb128()?),
            DuckDbPhysicalType::Int128 => {
                let upper = self.read_signed_leb128()?;
                let lower = self.read_unsigned_leb128()?;
                ((upper as i128) << 64) | i128::from(lower)
            }
            other => {
                return Err(StorageError::Backend(format!(
                    "DuckDB DECIMAL mapped to unexpected physical type {other:?}"
                )));
            }
        };
        Ok(DuckDbValue::Decimal {
            value,
            width,
            scale,
        })
    }

    pub(crate) fn read_duckdb_interval_value(&mut self) -> Result<DuckDbValue> {
        let months = if self.consume_optional_field(1)? {
            i32::try_from(self.read_signed_leb128()?).map_err(|_| {
                StorageError::Backend("DuckDB Value INTERVAL months exceeds i32".into())
            })?
        } else {
            0
        };
        let days = if self.consume_optional_field(2)? {
            i32::try_from(self.read_signed_leb128()?).map_err(|_| {
                StorageError::Backend("DuckDB Value INTERVAL days exceeds i32".into())
            })?
        } else {
            0
        };
        let micros = if self.consume_optional_field(3)? {
            self.read_signed_leb128()?
        } else {
            0
        };
        self.read_object_end()?;
        Ok(DuckDbValue::Interval {
            months,
            days,
            micros,
        })
    }

    pub(crate) fn read_duckdb_value_children(
        &mut self,
        child_types: &[DuckDbStructField],
    ) -> Result<Vec<DuckDbValue>> {
        self.read_property_begin(100, "children")?;
        let count = self.read_usize()?;
        if count != child_types.len() {
            return Err(StorageError::Backend(format!(
                "DuckDB nested Value child count mismatch: expected {}, got {count}",
                child_types.len()
            )));
        }
        let mut values = Vec::with_capacity(count);
        for child_type in child_types {
            values.push(self.read_duckdb_value_with_expected_type(Some(&child_type.logical_type))?);
        }
        self.read_object_end()?;
        Ok(values)
    }

    pub(crate) fn read_duckdb_repeated_value_children(
        &mut self,
        child_type: &DuckDbLogicalType,
        expected_count: Option<usize>,
    ) -> Result<Vec<DuckDbValue>> {
        self.read_property_begin(100, "children")?;
        let count = self.read_usize()?;
        if let Some(expected_count) = expected_count
            && count != expected_count
        {
            return Err(StorageError::Backend(format!(
                "DuckDB ARRAY Value child count mismatch: expected {expected_count}, got {count}"
            )));
        }
        let mut values = Vec::with_capacity(count);
        for _ in 0..count {
            values.push(self.read_duckdb_value_with_expected_type(Some(child_type))?);
        }
        self.read_object_end()?;
        Ok(values)
    }

    pub(crate) fn read_deleted_extra_type_modifiers(&mut self) -> Result<()> {
        if !self.consume_optional_field(102)? {
            return Ok(());
        }
        let count = self.read_usize()?;
        for _ in 0..count {
            let _ = self.read_duckdb_value()?;
        }
        Ok(())
    }

    pub(crate) fn read_extension_type_info(&mut self) -> Result<()> {
        if !self.consume_optional_field(103)? {
            return Ok(());
        }

        let present = self.read_u8()? != 0;
        if !present {
            return Ok(());
        }

        self.read_property_begin(100, "modifiers")?;
        let modifier_count = self.read_usize()?;
        for _ in 0..modifier_count {
            self.read_property_begin(100, "value")?;
            let _ = self.read_duckdb_value()?;
            let _ = self.read_optional_string_property(101, "label")?;
            self.read_object_end()?;
        }

        if self.consume_optional_field(101)? {
            let property_count = self.read_usize()?;
            for _ in 0..property_count {
                self.read_property_begin(0, "key")?;
                let _ = self.read_string()?;
                self.read_property_begin(1, "value")?;
                let _ = self.read_duckdb_value()?;
                self.read_object_end()?;
            }
        }

        self.read_object_end()
    }

    pub(crate) fn read_optional_data_chunk_vector_property(
        &mut self,
        field_id: u16,
        _tag: &str,
    ) -> Result<Vec<DuckDbDataChunk>> {
        if !self.consume_optional_field(field_id)? {
            return Ok(Vec::new());
        }
        let count = self.read_usize()?;
        let mut chunks = Vec::with_capacity(count);
        for _ in 0..count {
            chunks.push(self.read_data_chunk_wrapper()?);
        }
        Ok(chunks)
    }

    pub(crate) fn read_optional_u8_property(&mut self, field_id: u16, _tag: &str) -> Result<u8> {
        if !self.consume_optional_field(field_id)? {
            return Ok(0);
        }
        let value = self.read_unsigned_leb128()?;
        u8::try_from(value)
            .map_err(|_| StorageError::Backend(format!("DuckDB u8 property exceeds u8: {value}")))
    }

    pub(crate) fn read_optional_u32_property(&mut self, field_id: u16, _tag: &str) -> Result<u32> {
        if !self.consume_optional_field(field_id)? {
            return Ok(0);
        }
        let value = self.read_unsigned_leb128()?;
        u32::try_from(value)
            .map_err(|_| StorageError::Backend(format!("DuckDB u32 property exceeds u32: {value}")))
    }

    pub(crate) fn read_optional_logical_type_list_property(
        &mut self,
        field_id: u16,
        _tag: &str,
    ) -> Result<Vec<DuckDbLogicalType>> {
        if !self.consume_optional_field(field_id)? {
            return Ok(Vec::new());
        }
        let count = self.read_usize()?;
        let mut values = Vec::with_capacity(count);
        for _ in 0..count {
            values.push(self.read_logical_type()?);
        }
        Ok(values)
    }

    pub(crate) fn read_optional_child_type_list_property(
        &mut self,
        field_id: u16,
        _tag: &str,
    ) -> Result<Vec<DuckDbStructField>> {
        if !self.consume_optional_field(field_id)? {
            return Ok(Vec::new());
        }
        let count = self.read_usize()?;
        let mut children = Vec::with_capacity(count);
        for _ in 0..count {
            self.read_property_begin(0, "first")?;
            let name = self.read_string()?;
            self.read_property_begin(1, "second")?;
            let logical_type = self.read_logical_type()?;
            self.read_object_end()?;
            children.push(DuckDbStructField { name, logical_type });
        }
        Ok(children)
    }

    pub(crate) fn read_string_list(&mut self) -> Result<Vec<String>> {
        let count = self.read_usize()?;
        let mut values = Vec::with_capacity(count);
        for _ in 0..count {
            values.push(self.read_string()?);
        }
        Ok(values)
    }

    pub(crate) fn read_data_chunk_wrapper(&mut self) -> Result<DuckDbDataChunk> {
        let present = self.read_u8()? != 0;
        if !present {
            return Err(StorageError::Unsupported(
                "Quack null DataChunk response decoding is not supported".into(),
            ));
        }
        self.read_property_begin(300, "chunk")?;
        let chunk = self.read_data_chunk()?;
        self.read_object_end()?;
        Ok(chunk)
    }

    pub(crate) fn read_data_chunk(&mut self) -> Result<DuckDbDataChunk> {
        self.read_property_begin(100, "rows")?;
        let row_count = self.read_usize()?;

        self.read_property_begin(101, "types")?;
        let column_count = self.read_usize()?;
        let mut logical_types = Vec::with_capacity(column_count);
        for _ in 0..column_count {
            logical_types.push(self.read_logical_type()?);
        }

        self.read_property_begin(102, "columns")?;
        let serialized_columns = self.read_usize()?;
        if serialized_columns != logical_types.len() {
            return Err(StorageError::Backend(format!(
                "DuckDB DataChunk column count mismatch: {} types, {} columns",
                logical_types.len(),
                serialized_columns
            )));
        }

        let mut columns = Vec::with_capacity(serialized_columns);
        for logical_type in &logical_types {
            let column = self.read_vector(logical_type, row_count)?;
            self.read_object_end()?;
            columns.push(column);
        }
        self.read_object_end()?;

        let mut rows = Vec::with_capacity(row_count);
        for row_idx in 0..row_count {
            let mut row = Vec::with_capacity(columns.len());
            for column in &columns {
                row.push(column[row_idx].clone());
            }
            rows.push(row);
        }

        Ok(DuckDbDataChunk {
            column_types: logical_types
                .iter()
                .map(DuckDbLogicalType::type_name)
                .collect(),
            rows,
        })
    }

    pub(crate) fn read_vector(
        &mut self,
        logical_type: &DuckDbLogicalType,
        row_count: usize,
    ) -> Result<Vec<DuckDbValue>> {
        let vector_type = if self.consume_optional_field(90)? {
            self.read_unsigned_leb128()?
        } else {
            0
        };
        match vector_type {
            0 => self.read_flat_vector(logical_type, row_count),
            2 => {
                let mut value = self.read_vector(logical_type, 1)?;
                let value = value.pop().unwrap_or(DuckDbValue::Null);
                Ok(vec![value; row_count])
            }
            3 => self.read_dictionary_vector(logical_type, row_count),
            4 => self.read_sequence_vector(logical_type, row_count),
            other => Err(StorageError::Unsupported(format!(
                "Quack DuckDB compressed Vector response decoding for VectorType {other} is not supported"
            ))),
        }
    }

    pub(crate) fn read_dictionary_vector(
        &mut self,
        logical_type: &DuckDbLogicalType,
        row_count: usize,
    ) -> Result<Vec<DuckDbValue>> {
        let sel_vector = self.read_raw_blob_property(91, "sel_vector")?.to_vec();
        let expected_len = row_count.checked_mul(4).ok_or_else(|| {
            StorageError::Backend("DuckDB selection vector length overflow".into())
        })?;
        if sel_vector.len() != expected_len {
            return Err(StorageError::Backend(format!(
                "DuckDB selection vector byte length mismatch: expected {expected_len}, got {}",
                sel_vector.len()
            )));
        }
        self.read_property_begin(92, "dict_count")?;
        let dict_count = self.read_usize()?;
        let dictionary = self.read_vector(logical_type, dict_count)?;

        let mut values = Vec::with_capacity(row_count);
        for row_idx in 0..row_count {
            let selection =
                usize::try_from(read_u32_le(&sel_vector, row_idx * 4)?).map_err(|_| {
                    StorageError::Backend("DuckDB selection index exceeds usize".into())
                })?;
            let value = dictionary.get(selection).ok_or_else(|| {
                StorageError::Backend(format!(
                    "DuckDB selection index {selection} exceeds dictionary length {}",
                    dictionary.len()
                ))
            })?;
            values.push(value.clone());
        }
        Ok(values)
    }

    pub(crate) fn read_sequence_vector(
        &mut self,
        logical_type: &DuckDbLogicalType,
        row_count: usize,
    ) -> Result<Vec<DuckDbValue>> {
        self.read_property_begin(91, "seq_start")?;
        let start = self.read_signed_leb128()?;
        self.read_property_begin(92, "seq_increment")?;
        let increment = self.read_signed_leb128()?;
        let mut values = Vec::with_capacity(row_count);
        for row_idx in 0..row_count {
            let offset = i64::try_from(row_idx).map_err(|_| {
                StorageError::Backend("DuckDB sequence row index exceeds i64".into())
            })?;
            let raw = increment
                .checked_mul(offset)
                .and_then(|value| start.checked_add(value))
                .ok_or_else(|| StorageError::Backend("DuckDB sequence value overflow".into()))?;
            values.push(duckdb_sequence_value(logical_type, raw)?);
        }
        Ok(values)
    }

    pub(crate) fn read_flat_vector(
        &mut self,
        logical_type: &DuckDbLogicalType,
        row_count: usize,
    ) -> Result<Vec<DuckDbValue>> {
        if self.peek_field()? == 99 {
            return Err(StorageError::Unsupported(
                "Quack DuckDB geometry vector response decoding is not supported".into(),
            ));
        }

        self.read_property_begin(100, "has_validity_mask")?;
        let has_validity_mask = self.read_u8()? != 0;
        let validity = if has_validity_mask {
            let validity = self.read_raw_blob_property(101, "validity")?.to_vec();
            let expected_len = duckdb_validity_mask_size(row_count);
            if validity.len() != expected_len {
                return Err(StorageError::Backend(format!(
                    "DuckDB validity mask byte length mismatch: expected {expected_len}, got {}",
                    validity.len()
                )));
            }
            Some(validity)
        } else {
            None
        };

        let physical_type = duckdb_physical_type(logical_type)?;
        let mut values = match physical_type {
            DuckDbPhysicalType::Varchar => {
                self.read_string_vector_data(logical_type, row_count, validity.as_deref())?
            }
            DuckDbPhysicalType::Blob => self.read_blob_vector_data(row_count)?,
            DuckDbPhysicalType::Struct => self.read_struct_vector_data(logical_type, row_count)?,
            DuckDbPhysicalType::List => self.read_list_vector_data(logical_type, row_count)?,
            DuckDbPhysicalType::Array => self.read_array_vector_data(logical_type, row_count)?,
            _ => {
                let bytes = self.read_raw_blob_property(102, "data")?.to_vec();
                self.decode_fixed_width_vector(logical_type, physical_type, row_count, &bytes)?
            }
        };

        if let Some(validity) = validity {
            for (row_idx, value) in values.iter_mut().enumerate() {
                if !duckdb_validity_row_is_valid(&validity, row_idx) {
                    *value = DuckDbValue::Null;
                }
            }
        }

        Ok(values)
    }

    pub(crate) fn read_struct_vector_data(
        &mut self,
        logical_type: &DuckDbLogicalType,
        row_count: usize,
    ) -> Result<Vec<DuckDbValue>> {
        let children = logical_type.struct_children()?;
        self.read_property_begin(103, "children")?;
        let child_count = self.read_usize()?;
        if child_count != children.len() {
            return Err(StorageError::Backend(format!(
                "DuckDB struct child count mismatch: expected {}, got {child_count}",
                children.len()
            )));
        }

        let mut child_columns = Vec::with_capacity(child_count);
        for child in children {
            let values = self.read_vector(&child.logical_type, row_count)?;
            self.read_object_end()?;
            child_columns.push((&child.name, values));
        }

        let mut rows = Vec::with_capacity(row_count);
        for row_idx in 0..row_count {
            let mut fields = Vec::with_capacity(child_columns.len());
            for (name, values) in &child_columns {
                fields.push(((*name).clone(), values[row_idx].clone()));
            }
            rows.push(DuckDbValue::Struct(fields));
        }
        Ok(rows)
    }

    pub(crate) fn read_list_vector_data(
        &mut self,
        logical_type: &DuckDbLogicalType,
        row_count: usize,
    ) -> Result<Vec<DuckDbValue>> {
        let child_type = logical_type.list_child_type()?;
        self.read_property_begin(104, "list_size")?;
        let list_size = self.read_usize()?;

        self.read_property_begin(105, "entries")?;
        let entry_count = self.read_usize()?;
        if entry_count != row_count {
            return Err(StorageError::Backend(format!(
                "DuckDB list entry count mismatch: expected {row_count}, got {entry_count}"
            )));
        }
        let mut entries = Vec::with_capacity(entry_count);
        for _ in 0..entry_count {
            self.read_property_begin(100, "offset")?;
            let offset = self.read_usize()?;
            self.read_property_begin(101, "length")?;
            let length = self.read_usize()?;
            self.read_object_end()?;
            entries.push(DuckDbListEntry { offset, length });
        }

        self.read_property_begin(106, "child")?;
        let child_values = self.read_vector(child_type, list_size)?;
        self.read_object_end()?;

        let mut values = Vec::with_capacity(row_count);
        for entry in entries {
            let end = entry
                .offset
                .checked_add(entry.length)
                .ok_or_else(|| StorageError::Backend("DuckDB list entry range overflow".into()))?;
            let slice = child_values.get(entry.offset..end).ok_or_else(|| {
                StorageError::Backend(format!(
                    "DuckDB list entry range {}..{} exceeds child length {}",
                    entry.offset,
                    end,
                    child_values.len()
                ))
            })?;
            if logical_type.id == 102 {
                values.push(DuckDbValue::Map(duckdb_map_entries_from_structs(slice)?));
            } else {
                values.push(DuckDbValue::List(slice.to_vec()));
            }
        }
        Ok(values)
    }

    pub(crate) fn read_array_vector_data(
        &mut self,
        logical_type: &DuckDbLogicalType,
        row_count: usize,
    ) -> Result<Vec<DuckDbValue>> {
        let (child_type, array_size) = logical_type.array_child_type_and_size()?;
        self.read_property_begin(103, "array_size")?;
        let serialized_array_size = self.read_usize()?;
        if serialized_array_size != array_size {
            return Err(StorageError::Backend(format!(
                "DuckDB array size mismatch: expected {array_size}, got {serialized_array_size}"
            )));
        }
        let child_count = array_size
            .checked_mul(row_count)
            .ok_or_else(|| StorageError::Backend("DuckDB array child count overflow".into()))?;
        self.read_property_begin(104, "child")?;
        let child_values = self.read_vector(child_type, child_count)?;
        self.read_object_end()?;

        let mut rows = Vec::with_capacity(row_count);
        for row_idx in 0..row_count {
            let start = row_idx * array_size;
            let end = start + array_size;
            rows.push(DuckDbValue::Array(child_values[start..end].to_vec()));
        }
        Ok(rows)
    }

    pub(crate) fn read_string_vector_data(
        &mut self,
        _logical_type: &DuckDbLogicalType,
        row_count: usize,
        validity: Option<&[u8]>,
    ) -> Result<Vec<DuckDbValue>> {
        self.read_property_begin(102, "data")?;
        let count = self.read_usize()?;
        if count != row_count {
            return Err(StorageError::Backend(format!(
                "DuckDB string vector row count mismatch: expected {row_count}, got {count}"
            )));
        }
        let mut values = Vec::with_capacity(count);
        for row_idx in 0..count {
            let bytes = self.read_blob()?;
            if validity.is_some_and(|validity| !duckdb_validity_row_is_valid(validity, row_idx)) {
                values.push(DuckDbValue::Null);
                continue;
            }
            let value = std::str::from_utf8(bytes)
                .map(str::to_string)
                .map_err(|err| {
                    let prefix = &bytes[..bytes.len().min(16)];
                    StorageError::Backend(format!(
                        "invalid UTF-8 in Quack string vector row {row_idx} (len {}, prefix 0x{}): {err}",
                        bytes.len(),
                        crate::hex_encode_bytes(prefix)
                    ))
                })?;
            values.push(DuckDbValue::Utf8(value));
        }
        Ok(values)
    }

    pub(crate) fn read_blob_vector_data(&mut self, row_count: usize) -> Result<Vec<DuckDbValue>> {
        self.read_property_begin(102, "data")?;
        let count = self.read_usize()?;
        if count != row_count {
            return Err(StorageError::Backend(format!(
                "DuckDB blob vector row count mismatch: expected {row_count}, got {count}"
            )));
        }
        let mut values = Vec::with_capacity(count);
        for _ in 0..count {
            values.push(DuckDbValue::Blob(self.read_blob()?.to_vec()));
        }
        Ok(values)
    }

    pub(crate) fn decode_fixed_width_vector(
        &self,
        logical_type: &DuckDbLogicalType,
        physical_type: DuckDbPhysicalType,
        row_count: usize,
        bytes: &[u8],
    ) -> Result<Vec<DuckDbValue>> {
        let width = duckdb_physical_type_width(physical_type)?;
        let expected_len = width
            .checked_mul(row_count)
            .ok_or_else(|| StorageError::Backend("DuckDB vector byte length overflow".into()))?;
        if bytes.len() != expected_len {
            return Err(StorageError::Backend(format!(
                "DuckDB fixed-width vector byte length mismatch: expected {expected_len}, got {}",
                bytes.len()
            )));
        }

        let mut values = Vec::with_capacity(row_count);
        for row_idx in 0..row_count {
            let start = row_idx * width;
            let value = match logical_type.id {
                10 => DuckDbValue::Boolean(bytes[start] != 0),
                11 => DuckDbValue::Int8(bytes[start] as i8),
                12 => DuckDbValue::Int16(read_i16_le(bytes, start)?),
                13 => DuckDbValue::Int32(read_i32_le(bytes, start)?),
                14 => DuckDbValue::Int64(read_i64_le(bytes, start)?),
                15 => DuckDbValue::DateDays(read_i32_le(bytes, start)?),
                16 => DuckDbValue::TimeMicros(read_i64_le(bytes, start)?),
                17 => DuckDbValue::TimestampSeconds(read_i64_le(bytes, start)?),
                18 => DuckDbValue::TimestampMillis(read_i64_le(bytes, start)?),
                19 => DuckDbValue::TimestampMicros(read_i64_le(bytes, start)?),
                20 => DuckDbValue::TimestampNanos(read_i64_le(bytes, start)?),
                22 => DuckDbValue::Float32(read_f32_le(bytes, start)?),
                23 => DuckDbValue::Float64(read_f64_le(bytes, start)?),
                27 => DuckDbValue::Interval {
                    months: read_i32_le(bytes, start)?,
                    days: read_i32_le(bytes, start + 4)?,
                    micros: read_i64_le(bytes, start + 8)?,
                },
                28 => DuckDbValue::UInt8(bytes[start]),
                29 => DuckDbValue::UInt16(read_u16_le(bytes, start)?),
                30 => DuckDbValue::UInt32(read_u32_le(bytes, start)?),
                31 => DuckDbValue::UInt64(read_u64_le(bytes, start)?),
                32 => DuckDbValue::TimestampTzMicros(read_i64_le(bytes, start)?),
                34 => DuckDbValue::TimeTzMicros(read_i64_le(bytes, start)?),
                35 => DuckDbValue::TimeNanos(read_i64_le(bytes, start)?),
                49 => DuckDbValue::UInt128(read_duckdb_u128(bytes, start)?),
                50 => DuckDbValue::Int128(read_duckdb_i128(bytes, start)?),
                54 => DuckDbValue::Uuid(read_duckdb_uuid_bytes(bytes, start)?),
                104 => {
                    let index = match physical_type {
                        DuckDbPhysicalType::UInt8 => usize::from(bytes[start]),
                        DuckDbPhysicalType::UInt16 => usize::from(read_u16_le(bytes, start)?),
                        DuckDbPhysicalType::UInt32 => usize::try_from(read_u32_le(bytes, start)?)
                            .map_err(|_| {
                            StorageError::Backend("DuckDB enum index exceeds usize".into())
                        })?,
                        _ => {
                            return Err(StorageError::Backend(
                                "DuckDB ENUM vector used an unexpected physical type".into(),
                            ));
                        }
                    };
                    let values = logical_type.enum_values()?;
                    let value = values
                        .get(index)
                        .ok_or_else(|| {
                            StorageError::Backend(format!(
                                "DuckDB enum index {index} exceeds value count {}",
                                values.len()
                            ))
                        })?
                        .clone();
                    DuckDbValue::Utf8(value)
                }
                21 => {
                    let (width, scale) = logical_type.decimal_width_scale().ok_or_else(|| {
                        StorageError::Backend(
                            "DuckDB DECIMAL logical type is missing width/scale".into(),
                        )
                    })?;
                    let value = match physical_type {
                        DuckDbPhysicalType::Int16 => i128::from(read_i16_le(bytes, start)?),
                        DuckDbPhysicalType::Int32 => i128::from(read_i32_le(bytes, start)?),
                        DuckDbPhysicalType::Int64 => i128::from(read_i64_le(bytes, start)?),
                        DuckDbPhysicalType::Int128 => read_duckdb_i128(bytes, start)?,
                        _ => {
                            return Err(StorageError::Backend(
                                "DuckDB DECIMAL vector used an unexpected physical type".into(),
                            ));
                        }
                    };
                    DuckDbValue::Decimal {
                        value,
                        width,
                        scale,
                    }
                }
                other => {
                    return Err(StorageError::Unsupported(format!(
                        "Quack DuckDB DataChunk value decoding for LogicalTypeId {other} is not supported"
                    )));
                }
            };
            values.push(value);
        }
        Ok(values)
    }

    pub(crate) fn read_raw_blob_property(&mut self, field_id: u16, tag: &str) -> Result<&'a [u8]> {
        self.read_property_begin(field_id, tag)?;
        self.read_blob()
    }

    pub(crate) fn read_blob(&mut self) -> Result<&'a [u8]> {
        let len = self.read_usize()?;
        self.read_exact(len)
    }

    pub(crate) fn read_optional_bool_property(
        &mut self,
        field_id: u16,
        _tag: &str,
    ) -> Result<bool> {
        if !self.consume_optional_field(field_id)? {
            return Ok(false);
        }
        Ok(self.read_u8()? != 0)
    }

    pub(crate) fn read_optional_idx_property(
        &mut self,
        field_id: u16,
        tag: &str,
    ) -> Result<Option<u64>> {
        let value = self.read_idx_property(field_id, tag)?;
        Ok((value != DUCKDB_INVALID_OPTIONAL_IDX).then_some(value))
    }

    pub(crate) fn read_optional_idx_property_value(
        &mut self,
        field_id: u16,
        _tag: &str,
    ) -> Result<u64> {
        if !self.consume_optional_field(field_id)? {
            return Ok(0);
        }
        self.read_unsigned_leb128()
    }

    pub(crate) fn read_idx_property(&mut self, field_id: u16, tag: &str) -> Result<u64> {
        self.read_property_begin(field_id, tag)?;
        self.read_unsigned_leb128()
    }

    pub(crate) fn read_hugeint_property(&mut self, field_id: u16, tag: &str) -> Result<i128> {
        self.read_property_begin(field_id, tag)?;
        let upper = self.read_signed_leb128()?;
        let lower = self.read_unsigned_leb128()?;
        Ok(((upper as i128) << 64) | lower as i128)
    }

    pub(crate) fn read_message_type_property(
        &mut self,
        field_id: u16,
        tag: &str,
    ) -> Result<MessageType> {
        self.read_property_begin(field_id, tag)?;
        let value = self.read_unsigned_leb128()?;
        let value = u8::try_from(value)
            .map_err(|_| StorageError::Backend(format!("invalid Quack message type {value}")))?;
        MessageType::try_from(value)
    }

    pub(crate) fn read_property_begin(&mut self, expected_field_id: u16, tag: &str) -> Result<()> {
        let actual = self.next_field()?;
        if actual == expected_field_id {
            return Ok(());
        }
        Err(StorageError::Backend(format!(
            "failed to deserialize DuckDB property {tag}: expected field id {expected_field_id}, got {actual}"
        )))
    }

    pub(crate) fn read_object_end(&mut self) -> Result<()> {
        let position = self.position;
        let actual = self.next_field()?;
        if actual == DUCKDB_MESSAGE_TERMINATOR_FIELD_ID {
            return Ok(());
        }
        Err(StorageError::Backend(format!(
            "failed to deserialize DuckDB object at byte {position}: expected terminator, got field id {actual}"
        )))
    }

    pub(crate) fn consume_optional_field(&mut self, field_id: u16) -> Result<bool> {
        let actual = self.peek_field()?;
        if actual == field_id {
            self.consume_field()?;
            return Ok(true);
        }
        Ok(false)
    }

    pub(crate) fn peek_field(&mut self) -> Result<u16> {
        if let Some(field) = self.buffered_field {
            return Ok(field);
        }
        let field = self.read_field_id()?;
        self.buffered_field = Some(field);
        Ok(field)
    }

    pub(crate) fn consume_field(&mut self) -> Result<u16> {
        if let Some(field) = self.buffered_field.take() {
            return Ok(field);
        }
        self.read_field_id()
    }

    pub(crate) fn next_field(&mut self) -> Result<u16> {
        self.consume_field()
    }

    pub(crate) fn read_field_id(&mut self) -> Result<u16> {
        let bytes = self.read_exact(2)?;
        Ok(u16::from_le_bytes([bytes[0], bytes[1]]))
    }

    pub(crate) fn read_string(&mut self) -> Result<String> {
        let bytes = self.read_blob()?;
        std::str::from_utf8(bytes)
            .map(str::to_string)
            .map_err(|err| {
                let prefix = &bytes[..bytes.len().min(16)];
                StorageError::Backend(format!(
                    "invalid UTF-8 in Quack string (len {}, prefix 0x{}): {err}",
                    bytes.len(),
                    crate::hex_encode_bytes(prefix)
                ))
            })
    }

    pub(crate) fn read_usize(&mut self) -> Result<usize> {
        let value = self.read_unsigned_leb128()?;
        usize::try_from(value)
            .map_err(|_| StorageError::Backend("DuckDB integer value exceeds usize".into()))
    }

    pub(crate) fn read_unsigned_leb128(&mut self) -> Result<u64> {
        let mut result = 0u64;
        let mut shift = 0u32;
        for _ in 0..10 {
            let byte = self.read_u8()?;
            if shift >= 64 && (byte & 0x7f) != 0 {
                return Err(StorageError::Backend(
                    "failed to decode DuckDB unsigned LEB128".into(),
                ));
            }
            result |= u64::from(byte & 0x7f) << shift;
            if byte & 0x80 == 0 {
                return Ok(result);
            }
            shift += 7;
        }
        Err(StorageError::Backend(
            "failed to decode DuckDB unsigned LEB128".into(),
        ))
    }

    pub(crate) fn read_signed_leb128(&mut self) -> Result<i64> {
        let mut result = 0i64;
        let mut shift = 0u32;
        let mut byte;
        loop {
            byte = self.read_u8()?;
            result |= i64::from(byte & 0x7f) << shift;
            shift += 7;
            if byte & 0x80 == 0 {
                break;
            }
            if shift >= 64 {
                return Err(StorageError::Backend(
                    "failed to decode DuckDB signed LEB128".into(),
                ));
            }
        }
        if shift < 64 && (byte & 0x40) != 0 {
            result |= !0i64 << shift;
        }
        Ok(result)
    }

    pub(crate) fn read_u8(&mut self) -> Result<u8> {
        let bytes = self.read_exact(1)?;
        Ok(bytes[0])
    }

    pub(crate) fn read_exact(&mut self, len: usize) -> Result<&'a [u8]> {
        let end = self
            .position
            .checked_add(len)
            .ok_or_else(|| StorageError::Backend("Quack binary read overflow".into()))?;
        let bytes = self
            .input
            .get(self.position..end)
            .ok_or_else(|| StorageError::Backend("truncated Quack DuckDB binary message".into()))?;
        self.position = end;
        Ok(bytes)
    }
}

pub(crate) const HEADER_PROPERTIES: [DuckDbPropertyDescriptor; 3] = [
    DuckDbPropertyDescriptor::new(1, "type", DuckDbPropertyKind::MessageType, false),
    DuckDbPropertyDescriptor::new(2, "connection_id", DuckDbPropertyKind::String, true),
    DuckDbPropertyDescriptor::new(3, "client_query_id", DuckDbPropertyKind::OptionalIdx, false),
];
pub(crate) const CONNECTION_REQUEST_PROPERTIES: [DuckDbPropertyDescriptor; 5] = [
    DuckDbPropertyDescriptor::new(1, "auth_string", DuckDbPropertyKind::String, true),
    DuckDbPropertyDescriptor::new(2, "client_duckdb_version", DuckDbPropertyKind::String, true),
    DuckDbPropertyDescriptor::new(3, "client_platform", DuckDbPropertyKind::String, true),
    DuckDbPropertyDescriptor::new(
        4,
        "min_supported_quack_version",
        DuckDbPropertyKind::Idx,
        true,
    ),
    DuckDbPropertyDescriptor::new(
        5,
        "max_supported_quack_version",
        DuckDbPropertyKind::Idx,
        true,
    ),
];
pub(crate) const PREPARE_REQUEST_PROPERTIES: [DuckDbPropertyDescriptor; 1] =
    [DuckDbPropertyDescriptor::new(
        1,
        "sql_query",
        DuckDbPropertyKind::String,
        true,
    )];
pub(crate) const FETCH_REQUEST_PROPERTIES: [DuckDbPropertyDescriptor; 1] =
    [DuckDbPropertyDescriptor::new(
        1,
        "uuid",
        DuckDbPropertyKind::HugeInt,
        false,
    )];
pub(crate) const DISCONNECT_REQUEST_PROPERTIES: [DuckDbPropertyDescriptor; 0] = [];

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ConnectionRequest {
    pub header: MessageHeader,
    pub auth_string: String,
    pub client_duckdb_version: String,
    pub client_platform: String,
    pub min_supported_quack_version: u64,
    pub max_supported_quack_version: u64,
}

impl ConnectionRequest {
    pub fn new(auth_string: impl Into<String>) -> Self {
        Self {
            header: MessageHeader::new(MessageType::ConnectionRequest, ""),
            auth_string: auth_string.into(),
            client_duckdb_version: env!("CARGO_PKG_VERSION").to_string(),
            client_platform: format!("{}-{}", std::env::consts::OS, std::env::consts::ARCH),
            min_supported_quack_version: QUACK_VERSION,
            max_supported_quack_version: QUACK_VERSION,
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PrepareRequest {
    pub header: MessageHeader,
    pub sql_query: String,
}

impl PrepareRequest {
    pub fn new(connection_id: impl Into<String>, sql_query: impl Into<String>) -> Self {
        Self {
            header: MessageHeader::new(MessageType::PrepareRequest, connection_id),
            sql_query: sql_query.into(),
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct FetchRequest {
    pub header: MessageHeader,
    pub result_uuid: i128,
}

impl FetchRequest {
    pub fn new(connection_id: impl Into<String>, result_uuid: i128) -> Self {
        Self {
            header: MessageHeader::new(MessageType::FetchRequest, connection_id),
            result_uuid,
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DisconnectRequest {
    pub header: MessageHeader,
}

impl DisconnectRequest {
    pub fn new(connection_id: impl Into<String>) -> Self {
        Self {
            header: MessageHeader::new(MessageType::DisconnectMessage, connection_id),
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum QuackRequest {
    Connection(ConnectionRequest),
    Prepare(PrepareRequest),
    Fetch(FetchRequest),
    Disconnect(DisconnectRequest),
}

impl QuackRequest {
    pub fn header(&self) -> &MessageHeader {
        match self {
            Self::Connection(request) => &request.header,
            Self::Prepare(request) => &request.header,
            Self::Fetch(request) => &request.header,
            Self::Disconnect(request) => &request.header,
        }
    }

    pub fn message_type(&self) -> MessageType {
        self.header().message_type
    }

    pub fn expected_response_type(&self) -> MessageType {
        match self {
            Self::Connection(_) => MessageType::ConnectionResponse,
            Self::Prepare(_) => MessageType::PrepareResponse,
            Self::Fetch(_) => MessageType::FetchResponse,
            Self::Disconnect(_) => MessageType::SuccessResponse,
        }
    }

    pub fn with_client_query_id(mut self, client_query_id: u64) -> Self {
        match &mut self {
            Self::Connection(request) => request.header.client_query_id = Some(client_query_id),
            Self::Prepare(request) => request.header.client_query_id = Some(client_query_id),
            Self::Fetch(request) => request.header.client_query_id = Some(client_query_id),
            Self::Disconnect(request) => request.header.client_query_id = Some(client_query_id),
        }
        self
    }

    pub fn duckdb_header_descriptors() -> &'static [DuckDbPropertyDescriptor] {
        &HEADER_PROPERTIES
    }

    pub fn duckdb_body_descriptors(&self) -> &'static [DuckDbPropertyDescriptor] {
        match self {
            Self::Connection(_) => &CONNECTION_REQUEST_PROPERTIES,
            Self::Prepare(_) => &PREPARE_REQUEST_PROPERTIES,
            Self::Fetch(_) => &FETCH_REQUEST_PROPERTIES,
            Self::Disconnect(_) => &DISCONNECT_REQUEST_PROPERTIES,
        }
    }

    pub fn duckdb_message_plan(&self) -> DuckDbMessagePlan {
        DuckDbMessagePlan {
            header: self.duckdb_header_properties(),
            body: self.duckdb_body_properties(),
        }
    }

    fn duckdb_header_properties(&self) -> Vec<DuckDbProperty> {
        let header = self.header();
        vec![
            DuckDbProperty::new(
                HEADER_PROPERTIES[0],
                DuckDbPropertyValue::MessageType(header.message_type),
            ),
            DuckDbProperty::new(
                HEADER_PROPERTIES[1],
                DuckDbPropertyValue::String(header.connection_id.clone()),
            ),
            DuckDbProperty::new(
                HEADER_PROPERTIES[2],
                DuckDbPropertyValue::OptionalIdx(header.client_query_id),
            ),
        ]
    }

    fn duckdb_body_properties(&self) -> Vec<DuckDbProperty> {
        match self {
            Self::Connection(request) => vec![
                DuckDbProperty::new(
                    CONNECTION_REQUEST_PROPERTIES[0],
                    DuckDbPropertyValue::String(request.auth_string.clone()),
                ),
                DuckDbProperty::new(
                    CONNECTION_REQUEST_PROPERTIES[1],
                    DuckDbPropertyValue::String(request.client_duckdb_version.clone()),
                ),
                DuckDbProperty::new(
                    CONNECTION_REQUEST_PROPERTIES[2],
                    DuckDbPropertyValue::String(request.client_platform.clone()),
                ),
                DuckDbProperty::new(
                    CONNECTION_REQUEST_PROPERTIES[3],
                    DuckDbPropertyValue::Idx(request.min_supported_quack_version),
                ),
                DuckDbProperty::new(
                    CONNECTION_REQUEST_PROPERTIES[4],
                    DuckDbPropertyValue::Idx(request.max_supported_quack_version),
                ),
            ],
            Self::Prepare(request) => vec![DuckDbProperty::new(
                PREPARE_REQUEST_PROPERTIES[0],
                DuckDbPropertyValue::String(request.sql_query.clone()),
            )],
            Self::Fetch(request) => vec![DuckDbProperty::new(
                FETCH_REQUEST_PROPERTIES[0],
                DuckDbPropertyValue::HugeInt(request.result_uuid),
            )],
            Self::Disconnect(_) => Vec::new(),
        }
    }

    pub fn validate_response(&self, response: &QuackResponse) -> Result<()> {
        if let QuackResponse::Error(error) = response {
            return Err(StorageError::Backend(error.message.clone()));
        }
        let expected = self.expected_response_type();
        let actual = response.message_type();
        if actual != expected {
            return Err(StorageError::Backend(format!(
                "expected {} message, got {} instead",
                expected.as_str(),
                actual.as_str()
            )));
        }
        Ok(())
    }

    pub fn to_duckdb_binary(&self) -> Result<Vec<u8>> {
        let plan = self.duckdb_message_plan();
        let mut writer = DuckDbBinaryWriter::default();
        writer.write_object(&plan.header)?;
        writer.write_object(&plan.body)?;
        Ok(writer.finish())
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ConnectionResponse {
    pub header: MessageHeader,
    pub server_duckdb_version: String,
    pub server_platform: String,
    pub quack_version: u64,
}

impl ConnectionResponse {
    pub fn new(
        connection_id: impl Into<String>,
        server_duckdb_version: impl Into<String>,
        server_platform: impl Into<String>,
        quack_version: u64,
    ) -> Self {
        Self {
            header: MessageHeader::new(MessageType::ConnectionResponse, connection_id),
            server_duckdb_version: server_duckdb_version.into(),
            server_platform: server_platform.into(),
            quack_version,
        }
    }
}

#[derive(Debug, Clone, PartialEq)]
pub enum DuckDbValue {
    Null,
    Boolean(bool),
    Int8(i8),
    Int16(i16),
    Int32(i32),
    Int64(i64),
    Int128(i128),
    UInt8(u8),
    UInt16(u16),
    UInt32(u32),
    UInt64(u64),
    UInt128(u128),
    Float32(f32),
    Float64(f64),
    Decimal { value: i128, width: u8, scale: u8 },
    Utf8(String),
    Blob(Vec<u8>),
    Uuid([u8; 16]),
    Type(String),
    Struct(Vec<(String, DuckDbValue)>),
    List(Vec<DuckDbValue>),
    Map(Vec<(DuckDbValue, DuckDbValue)>),
    Array(Vec<DuckDbValue>),
    Interval { months: i32, days: i32, micros: i64 },
    DateDays(i32),
    TimeMicros(i64),
    TimeNanos(i64),
    TimestampSeconds(i64),
    TimestampMillis(i64),
    TimestampMicros(i64),
    TimestampNanos(i64),
    TimestampTzMicros(i64),
    TimeTzMicros(i64),
}

#[derive(Debug, Clone, PartialEq)]
pub struct DuckDbDataChunk {
    pub column_types: Vec<String>,
    pub rows: Vec<Vec<DuckDbValue>>,
}

#[derive(Debug, Clone, PartialEq)]
pub struct QuackQueryResult {
    pub column_names: Vec<String>,
    pub column_types: Vec<String>,
    pub rows: Vec<Vec<DuckDbValue>>,
}

impl QuackQueryResult {
    pub(crate) fn empty() -> Self {
        Self {
            column_names: Vec::new(),
            column_types: Vec::new(),
            rows: Vec::new(),
        }
    }

    pub(crate) fn from_chunks(
        column_names: Vec<String>,
        column_types: Vec<String>,
        chunks: Vec<DuckDbDataChunk>,
    ) -> Result<Self> {
        if !column_names.is_empty() && column_names.len() != column_types.len() {
            return Err(StorageError::Backend(format!(
                "Quack result metadata mismatch: {} names, {} types",
                column_names.len(),
                column_types.len()
            )));
        }

        let mut rows = Vec::new();
        for chunk in chunks {
            if chunk.column_types.len() != column_types.len() {
                return Err(StorageError::Backend(format!(
                    "Quack result chunk column mismatch: {} metadata columns, {} chunk columns",
                    column_types.len(),
                    chunk.column_types.len()
                )));
            }
            for row in chunk.rows {
                if row.len() != column_types.len() {
                    return Err(StorageError::Backend(format!(
                        "Quack result row column mismatch: {} metadata columns, {} row columns",
                        column_types.len(),
                        row.len()
                    )));
                }
                rows.push(row);
            }
        }
        Ok(Self {
            column_names,
            column_types,
            rows,
        })
    }

    pub(crate) fn typed_rows(&self) -> impl Iterator<Item = QuackRow<'_>> {
        self.rows.iter().map(|row| QuackRow { values: row })
    }
}

#[derive(Debug, Clone, Copy)]
pub(crate) struct QuackRow<'a> {
    pub(crate) values: &'a [DuckDbValue],
}

impl<'a> QuackRow<'a> {
    pub(crate) fn value(&self, idx: usize) -> Result<&'a DuckDbValue> {
        self.values
            .get(idx)
            .ok_or_else(|| StorageError::Backend(format!("Quack result row missing column {idx}")))
    }

    pub(crate) fn get_i64(&self, idx: usize) -> Result<i64> {
        self.opt_i64(idx)?.ok_or_else(|| {
            StorageError::Backend(format!(
                "Quack result column {idx} is NULL, expected BIGINT"
            ))
        })
    }

    pub(crate) fn get_timestamp_micros(&self, idx: usize) -> Result<i64> {
        match self.value(idx)? {
            DuckDbValue::Utf8(value) => {
                if let Ok(micros) = value.trim().parse::<i64>() {
                    return Ok(micros);
                }
                quack_parse_timestamp_micros(value).ok_or_else(|| {
                    StorageError::Backend(format!(
                        "Quack result column {idx} has invalid timestamp value {value:?}"
                    ))
                })
            }
            _ => self.get_i64(idx),
        }
    }

    pub(crate) fn opt_i64(&self, idx: usize) -> Result<Option<i64>> {
        let value = match self.value(idx)? {
            DuckDbValue::Null => return Ok(None),
            DuckDbValue::Int8(value) => i64::from(*value),
            DuckDbValue::Int16(value) => i64::from(*value),
            DuckDbValue::Int32(value) => i64::from(*value),
            DuckDbValue::Int64(value) => *value,
            DuckDbValue::UInt8(value) => i64::from(*value),
            DuckDbValue::UInt16(value) => i64::from(*value),
            DuckDbValue::UInt32(value) => i64::from(*value),
            DuckDbValue::UInt64(value) => i64::try_from(*value).map_err(|_| {
                StorageError::Backend(format!(
                    "Quack result column {idx} unsigned value exceeds i64"
                ))
            })?,
            DuckDbValue::DateDays(value) => i64::from(*value),
            DuckDbValue::TimeMicros(value)
            | DuckDbValue::TimeNanos(value)
            | DuckDbValue::TimestampSeconds(value)
            | DuckDbValue::TimestampMillis(value)
            | DuckDbValue::TimestampMicros(value)
            | DuckDbValue::TimestampNanos(value)
            | DuckDbValue::TimestampTzMicros(value)
            | DuckDbValue::TimeTzMicros(value) => *value,
            other => {
                return Err(StorageError::Backend(format!(
                    "Quack result column {idx} expected integer-compatible value, got {other:?}"
                )));
            }
        };
        Ok(Some(value))
    }

    pub(crate) fn get_string(&self, idx: usize) -> Result<String> {
        self.opt_string(idx)?.ok_or_else(|| {
            StorageError::Backend(format!(
                "Quack result column {idx} is NULL, expected VARCHAR"
            ))
        })
    }

    pub(crate) fn opt_string(&self, idx: usize) -> Result<Option<String>> {
        match self.value(idx)? {
            DuckDbValue::Null => Ok(None),
            DuckDbValue::Utf8(value) => Ok(Some(value.clone())),
            DuckDbValue::Uuid(value) => format_uuid_bytes(value, "Quack UUID").map(Some),
            other => Err(StorageError::Backend(format!(
                "Quack result column {idx} expected VARCHAR, got {other:?}"
            ))),
        }
    }

    pub(crate) fn get_bool(&self, idx: usize) -> Result<bool> {
        self.opt_bool(idx)?.ok_or_else(|| {
            StorageError::Backend(format!(
                "Quack result column {idx} is NULL, expected BOOLEAN"
            ))
        })
    }

    pub(crate) fn opt_bool(&self, idx: usize) -> Result<Option<bool>> {
        let value = match self.value(idx)? {
            DuckDbValue::Null => Ok(None),
            DuckDbValue::Boolean(value) => Ok(Some(*value)),
            DuckDbValue::Int8(value) => Ok(Some(quack_integer_bool(idx, i64::from(*value))?)),
            DuckDbValue::Int16(value) => Ok(Some(quack_integer_bool(idx, i64::from(*value))?)),
            DuckDbValue::Int32(value) => Ok(Some(quack_integer_bool(idx, i64::from(*value))?)),
            DuckDbValue::Int64(value) => Ok(Some(quack_integer_bool(idx, *value)?)),
            DuckDbValue::UInt8(value) => Ok(Some(quack_integer_bool(idx, i64::from(*value))?)),
            DuckDbValue::UInt16(value) => Ok(Some(quack_integer_bool(idx, i64::from(*value))?)),
            DuckDbValue::UInt32(value) => Ok(Some(quack_integer_bool(idx, i64::from(*value))?)),
            DuckDbValue::UInt64(value) => {
                let value = i64::try_from(*value).map_err(|_| {
                    StorageError::Backend(format!(
                        "Quack result column {idx} integer BOOLEAN value exceeds i64"
                    ))
                })?;
                Ok(Some(quack_integer_bool(idx, value)?))
            }
            other => Err(StorageError::Backend(format!(
                "Quack result column {idx} expected BOOLEAN, got {other:?}"
            ))),
        }?;
        Ok(value)
    }
}

#[derive(Debug, Clone, PartialEq)]
pub(crate) struct DuckDbLogicalType {
    id: u64,
    name: &'static str,
    extra_info: Option<DuckDbExtraTypeInfo>,
}

impl DuckDbLogicalType {
    fn type_name(&self) -> String {
        match (self.id, self.extra_info.as_ref()) {
            (21, Some(DuckDbExtraTypeInfo::Decimal { width, scale, .. })) => {
                format!("DECIMAL({width},{scale})")
            }
            (25, Some(DuckDbExtraTypeInfo::String { collation, .. })) if !collation.is_empty() => {
                format!("VARCHAR COLLATE {collation}")
            }
            (101, Some(DuckDbExtraTypeInfo::List { child_type, .. })) => {
                format!("{}[]", child_type.type_name())
            }
            (102, Some(DuckDbExtraTypeInfo::List { child_type, .. })) => {
                if let Some((key, value)) = child_type.map_key_value_types() {
                    format!("MAP({}, {})", key.type_name(), value.type_name())
                } else {
                    self.name.to_string()
                }
            }
            (100, Some(DuckDbExtraTypeInfo::Struct { children, .. })) => {
                let fields = children
                    .iter()
                    .map(|child| format!("{} {}", child.name, child.logical_type.type_name()))
                    .collect::<Vec<_>>()
                    .join(", ");
                format!("STRUCT({fields})")
            }
            (107, Some(DuckDbExtraTypeInfo::Struct { children, .. })) => {
                let fields = children
                    .iter()
                    .map(|child| format!("{} {}", child.name, child.logical_type.type_name()))
                    .collect::<Vec<_>>()
                    .join(", ");
                format!("UNION({fields})")
            }
            (
                108,
                Some(DuckDbExtraTypeInfo::Array {
                    child_type, size, ..
                }),
            ) => {
                format!("{}[{size}]", child_type.type_name())
            }
            (104, Some(DuckDbExtraTypeInfo::Enum { values, .. })) => {
                let values = values
                    .iter()
                    .map(|value| ducklake_sql_string_literal(value))
                    .collect::<Vec<_>>()
                    .join(", ");
                format!("ENUM({values})")
            }
            (38, Some(DuckDbExtraTypeInfo::IntegerLiteral { constant_value, .. })) => {
                duckdb_value_type_name(constant_value).to_string()
            }
            (4, Some(DuckDbExtraTypeInfo::Unbound { type_name, .. })) => type_name.clone(),
            (
                105,
                Some(DuckDbExtraTypeInfo::AggregateState {
                    function_name,
                    return_type,
                    bound_argument_types,
                    ..
                }),
            ) => {
                let args = bound_argument_types
                    .iter()
                    .map(DuckDbLogicalType::type_name)
                    .collect::<Vec<_>>()
                    .join(", ");
                format!(
                    "AGGREGATE_STATE<{function_name}({args})::{}>",
                    return_type.type_name()
                )
            }
            (5, Some(DuckDbExtraTypeInfo::Template { name, .. })) if !name.is_empty() => {
                name.clone()
            }
            (60, Some(DuckDbExtraTypeInfo::Geo { crs_definition, .. }))
                if !crs_definition.is_empty() =>
            {
                format!("GEOMETRY('{}')", crs_definition.replace('\'', "''"))
            }
            _ => self.name.to_string(),
        }
    }

    fn decimal_width_scale(&self) -> Option<(u8, u8)> {
        match self.extra_info.as_ref() {
            Some(DuckDbExtraTypeInfo::Decimal { width, scale, .. }) => Some((*width, *scale)),
            _ => None,
        }
    }

    fn map_key_value_types(&self) -> Option<(&DuckDbLogicalType, &DuckDbLogicalType)> {
        let Some(DuckDbExtraTypeInfo::Struct { children, .. }) = self.extra_info.as_ref() else {
            return None;
        };
        let key = children.iter().find(|child| child.name == "key")?;
        let value = children.iter().find(|child| child.name == "value")?;
        Some((&key.logical_type, &value.logical_type))
    }

    fn struct_children(&self) -> Result<&[DuckDbStructField]> {
        match self.extra_info.as_ref() {
            Some(DuckDbExtraTypeInfo::Struct { children, .. }) => Ok(children),
            _ => Err(StorageError::Backend(format!(
                "DuckDB {} logical type is missing struct children",
                self.name
            ))),
        }
    }

    fn list_child_type(&self) -> Result<&DuckDbLogicalType> {
        match self.extra_info.as_ref() {
            Some(DuckDbExtraTypeInfo::List { child_type, .. }) => Ok(child_type),
            _ => Err(StorageError::Backend(format!(
                "DuckDB {} logical type is missing list child type",
                self.name
            ))),
        }
    }

    fn array_child_type_and_size(&self) -> Result<(&DuckDbLogicalType, usize)> {
        match self.extra_info.as_ref() {
            Some(DuckDbExtraTypeInfo::Array {
                child_type, size, ..
            }) => {
                let size = usize::try_from(*size).map_err(|_| {
                    StorageError::Backend(format!("DuckDB ARRAY size exceeds usize: {size}"))
                })?;
                Ok((child_type, size))
            }
            _ => Err(StorageError::Backend(format!(
                "DuckDB {} logical type is missing array child type",
                self.name
            ))),
        }
    }

    fn enum_values(&self) -> Result<&[String]> {
        match self.extra_info.as_ref() {
            Some(DuckDbExtraTypeInfo::Enum { values, .. }) => Ok(values),
            _ => Err(StorageError::Backend(
                "DuckDB ENUM logical type is missing values".into(),
            )),
        }
    }
}

fn duckdb_unbound_type_name(
    catalog: &str,
    schema: &str,
    name: &str,
    modifiers: &[DuckDbValue],
) -> String {
    let mut parts = Vec::new();
    if !catalog.is_empty() {
        parts.push(catalog);
    }
    if !schema.is_empty() {
        parts.push(schema);
    }
    if !name.is_empty() {
        parts.push(name);
    }
    let mut type_name = if parts.is_empty() {
        "UNBOUND".to_string()
    } else {
        parts.join(".")
    };
    if !modifiers.is_empty() {
        let rendered = modifiers
            .iter()
            .map(duckdb_value_literal)
            .collect::<Vec<_>>()
            .join(", ");
        type_name.push('(');
        type_name.push_str(&rendered);
        type_name.push(')');
    }
    type_name
}

#[derive(Debug, Clone, PartialEq)]
pub(crate) struct DuckDbParsedExpressionHeader {
    expression_class: u64,
    expression_type: u64,
    alias: String,
}

#[derive(Debug, Clone, PartialEq)]
pub(crate) struct DuckDbParsedTypeExpression {
    alias: String,
    catalog: String,
    schema: String,
    type_name: String,
    children: Vec<DuckDbParsedExpression>,
}

#[derive(Debug, Clone, PartialEq)]
pub(crate) enum DuckDbParsedExpression {
    Type(DuckDbParsedTypeExpression),
    Constant {
        alias: String,
        value: DuckDbValue,
    },
    ColumnRef {
        alias: String,
        column_names: Vec<String>,
    },
    Comparison {
        alias: String,
        expression_type: u64,
        left: Option<Box<DuckDbParsedExpression>>,
        right: Option<Box<DuckDbParsedExpression>>,
    },
    Conjunction {
        alias: String,
        expression_type: u64,
        children: Vec<DuckDbParsedExpression>,
    },
    Function {
        alias: String,
        catalog: String,
        schema: String,
        function_name: String,
        children: Vec<DuckDbParsedExpression>,
        filter: Option<Box<DuckDbParsedExpression>>,
        distinct: bool,
        is_operator: bool,
        export_state: bool,
    },
    Operator {
        alias: String,
        expression_type: u64,
        children: Vec<DuckDbParsedExpression>,
    },
    Cast {
        alias: String,
        child: Option<Box<DuckDbParsedExpression>>,
        cast_type: DuckDbLogicalType,
        try_cast: bool,
    },
    Default {
        alias: String,
    },
    Parameter {
        alias: String,
        identifier: String,
    },
    PositionalReference {
        alias: String,
        index: u64,
    },
    Between {
        alias: String,
        negated: bool,
        input: Option<Box<DuckDbParsedExpression>>,
        lower: Option<Box<DuckDbParsedExpression>>,
        upper: Option<Box<DuckDbParsedExpression>>,
    },
    Case {
        alias: String,
        case_checks: Vec<DuckDbParsedCaseCheck>,
        else_expr: Option<Box<DuckDbParsedExpression>>,
    },
    Collate {
        alias: String,
        child: Option<Box<DuckDbParsedExpression>>,
        collation: String,
    },
    LambdaRef {
        alias: String,
        lambda_idx: u64,
        column_name: String,
    },
}

#[derive(Debug, Clone, PartialEq)]
pub(crate) struct DuckDbParsedCaseCheck {
    when_expr: Option<Box<DuckDbParsedExpression>>,
    then_expr: Option<Box<DuckDbParsedExpression>>,
}

impl DuckDbParsedExpression {
    fn alias(&self) -> &str {
        match self {
            Self::Type(expression) => &expression.alias,
            Self::Constant { alias, .. } => alias,
            Self::ColumnRef { alias, .. } => alias,
            Self::Comparison { alias, .. } => alias,
            Self::Conjunction { alias, .. } => alias,
            Self::Function { alias, .. } => alias,
            Self::Operator { alias, .. } => alias,
            Self::Cast { alias, .. } => alias,
            Self::Default { alias } => alias,
            Self::Parameter { alias, .. } => alias,
            Self::PositionalReference { alias, .. } => alias,
            Self::Between { alias, .. } => alias,
            Self::Case { alias, .. } => alias,
            Self::Collate { alias, .. } => alias,
            Self::LambdaRef { alias, .. } => alias,
        }
    }
}

fn duckdb_parsed_type_expression_name(expression: &DuckDbParsedTypeExpression) -> String {
    let qualified = duckdb_qualified_type_name(
        &expression.catalog,
        &expression.schema,
        &expression.type_name,
    );

    if expression.catalog.is_empty() && expression.schema.is_empty() {
        if expression.type_name.eq_ignore_ascii_case("LIST") && expression.children.len() == 1 {
            return format!(
                "{}[]",
                duckdb_parsed_type_parameter_name(&expression.children[0])
            );
        }
        if expression.type_name.eq_ignore_ascii_case("ARRAY") && expression.children.len() == 2 {
            return format!(
                "{}[{}]",
                duckdb_parsed_type_parameter_name(&expression.children[0]),
                duckdb_parsed_type_parameter_name(&expression.children[1])
            );
        }
        if expression.type_name.eq_ignore_ascii_case("STRUCT") {
            return duckdb_parsed_type_fields("STRUCT", &expression.children);
        }
        if expression.type_name.eq_ignore_ascii_case("UNION") {
            return duckdb_parsed_type_fields("UNION", &expression.children);
        }
        if expression.type_name.eq_ignore_ascii_case("MAP") && expression.children.len() == 2 {
            return format!(
                "MAP({}, {})",
                duckdb_parsed_type_parameter_name(&expression.children[0]),
                duckdb_parsed_type_parameter_name(&expression.children[1])
            );
        }
        if expression.type_name.eq_ignore_ascii_case("INTERVAL") && !expression.children.is_empty()
        {
            return "INTERVAL".to_string();
        }
    }

    if expression.children.is_empty() {
        qualified
    } else {
        let rendered = expression
            .children
            .iter()
            .map(duckdb_parsed_type_parameter_name)
            .collect::<Vec<_>>()
            .join(", ");
        format!("{qualified}({rendered})")
    }
}

fn duckdb_qualified_type_name(catalog: &str, schema: &str, name: &str) -> String {
    let mut parts = Vec::new();
    if !catalog.is_empty() {
        parts.push(catalog);
    }
    if !schema.is_empty() {
        parts.push(schema);
    }
    if !name.is_empty() {
        parts.push(name);
    }
    if parts.is_empty() {
        "UNBOUND".to_string()
    } else {
        parts.join(".")
    }
}

fn duckdb_parsed_type_fields(label: &str, children: &[DuckDbParsedExpression]) -> String {
    if children.is_empty() {
        return label.to_string();
    }
    let rendered = children
        .iter()
        .map(|child| {
            let child_type = duckdb_parsed_type_parameter_name(child);
            let alias = child.alias();
            if alias.is_empty() {
                child_type
            } else {
                format!("{alias} {child_type}")
            }
        })
        .collect::<Vec<_>>()
        .join(", ");
    format!("{label}({rendered})")
}

fn duckdb_parsed_type_parameter_name(parameter: &DuckDbParsedExpression) -> String {
    duckdb_parsed_expression_sql(parameter)
}

fn duckdb_parsed_expression_sql(expression: &DuckDbParsedExpression) -> String {
    match expression {
        DuckDbParsedExpression::Type(expression) => duckdb_parsed_type_expression_name(expression),
        DuckDbParsedExpression::Constant { value, .. } => duckdb_value_literal(value),
        DuckDbParsedExpression::ColumnRef { column_names, .. } => {
            if column_names.is_empty() {
                String::new()
            } else {
                column_names.join(".")
            }
        }
        DuckDbParsedExpression::Comparison {
            expression_type,
            left,
            right,
            ..
        } => {
            let left = duckdb_optional_expression_sql(left);
            let right = duckdb_optional_expression_sql(right);
            match *expression_type {
                25 => format!("{left} = {right}"),
                26 => format!("{left} != {right}"),
                27 => format!("{left} < {right}"),
                28 => format!("{left} > {right}"),
                29 => format!("{left} <= {right}"),
                30 => format!("{left} >= {right}"),
                35 => format!("{left} IN ({right})"),
                36 => format!("{left} NOT IN ({right})"),
                37 => format!("{left} IS DISTINCT FROM {right}"),
                40 => format!("{left} IS NOT DISTINCT FROM {right}"),
                other => format!("COMPARE_{other}({left}, {right})"),
            }
        }
        DuckDbParsedExpression::Conjunction {
            expression_type,
            children,
            ..
        } => {
            let separator = match *expression_type {
                50 => " AND ",
                51 => " OR ",
                _ => ", ",
            };
            children
                .iter()
                .map(|child| format!("({})", duckdb_parsed_expression_sql(child)))
                .collect::<Vec<_>>()
                .join(separator)
        }
        DuckDbParsedExpression::Function {
            catalog,
            schema,
            function_name,
            children,
            filter,
            distinct,
            is_operator,
            export_state,
            ..
        } => {
            let mut name = duckdb_qualified_type_name(catalog, schema, function_name);
            if *is_operator {
                name = format!("OPERATOR {name}");
            }
            let mut args = children
                .iter()
                .map(duckdb_parsed_expression_sql)
                .collect::<Vec<_>>();
            if *distinct {
                if let Some(first) = args.first_mut() {
                    *first = format!("DISTINCT {first}");
                } else {
                    args.push("DISTINCT".to_string());
                }
            }
            let mut rendered = format!("{name}({})", args.join(", "));
            if let Some(filter) = filter {
                rendered.push_str(" FILTER (WHERE ");
                rendered.push_str(&duckdb_parsed_expression_sql(filter));
                rendered.push(')');
            }
            if *export_state {
                rendered.push_str(" EXPORT_STATE");
            }
            rendered
        }
        DuckDbParsedExpression::Operator {
            expression_type,
            children,
            ..
        } => duckdb_operator_expression_sql(*expression_type, children),
        DuckDbParsedExpression::Cast {
            child,
            cast_type,
            try_cast,
            ..
        } => {
            let function = if *try_cast { "TRY_CAST" } else { "CAST" };
            format!(
                "{function}({} AS {})",
                duckdb_optional_expression_sql(child),
                cast_type.type_name()
            )
        }
        DuckDbParsedExpression::Default { .. } => "DEFAULT".to_string(),
        DuckDbParsedExpression::Parameter { identifier, .. } => {
            if identifier.is_empty() {
                "?".to_string()
            } else {
                format!("${identifier}")
            }
        }
        DuckDbParsedExpression::PositionalReference { index, .. } => format!("#{index}"),
        DuckDbParsedExpression::Between {
            negated,
            input,
            lower,
            upper,
            ..
        } => {
            let not = if *negated { " NOT" } else { "" };
            format!(
                "{}{not} BETWEEN {} AND {}",
                duckdb_optional_expression_sql(input),
                duckdb_optional_expression_sql(lower),
                duckdb_optional_expression_sql(upper)
            )
        }
        DuckDbParsedExpression::Case {
            case_checks,
            else_expr,
            ..
        } => {
            let mut rendered = "CASE".to_string();
            for check in case_checks {
                rendered.push_str(" WHEN ");
                rendered.push_str(&duckdb_optional_expression_sql(&check.when_expr));
                rendered.push_str(" THEN ");
                rendered.push_str(&duckdb_optional_expression_sql(&check.then_expr));
            }
            if let Some(else_expr) = else_expr {
                rendered.push_str(" ELSE ");
                rendered.push_str(&duckdb_parsed_expression_sql(else_expr));
            }
            rendered.push_str(" END");
            rendered
        }
        DuckDbParsedExpression::Collate {
            child, collation, ..
        } => format!(
            "{} COLLATE {}",
            duckdb_optional_expression_sql(child),
            collation
        ),
        DuckDbParsedExpression::LambdaRef {
            lambda_idx,
            column_name,
            ..
        } => {
            if column_name.is_empty() {
                format!("lambda_ref_{lambda_idx}")
            } else {
                column_name.clone()
            }
        }
    }
}

fn duckdb_optional_expression_sql(expression: &Option<Box<DuckDbParsedExpression>>) -> String {
    expression
        .as_deref()
        .map(duckdb_parsed_expression_sql)
        .unwrap_or_else(|| "NULL".to_string())
}

fn duckdb_operator_expression_sql(
    expression_type: u64,
    children: &[DuckDbParsedExpression],
) -> String {
    let args = children
        .iter()
        .map(duckdb_parsed_expression_sql)
        .collect::<Vec<_>>();
    match expression_type {
        13 if args.len() == 1 => format!("NOT {}", args[0]),
        14 if args.len() == 1 => format!("{} IS NULL", args[0]),
        15 if args.len() == 1 => format!("{} IS NOT NULL", args[0]),
        151 if args.len() == 2 => format!("NULLIF({}, {})", args[0], args[1]),
        152 => format!("COALESCE({})", args.join(", ")),
        153 if args.len() == 2 => format!("{}[{}]", args[0], args[1]),
        154 if args.len() == 3 => format!("{}[{}:{}]", args[0], args[1], args[2]),
        155 if args.len() == 2 => format!("{}.{}", args[0], args[1]),
        156 => format!("[{}]", args.join(", ")),
        158 if args.len() == 1 => format!("TRY({})", args[0]),
        other => format!("OPERATOR_{other}({})", args.join(", ")),
    }
}

fn duckdb_value_literal(value: &DuckDbValue) -> String {
    match value {
        DuckDbValue::Null => "NULL".to_string(),
        DuckDbValue::Boolean(value) => value.to_string(),
        DuckDbValue::Int8(value) => value.to_string(),
        DuckDbValue::Int16(value) => value.to_string(),
        DuckDbValue::Int32(value) => value.to_string(),
        DuckDbValue::Int64(value) => value.to_string(),
        DuckDbValue::Int128(value) => value.to_string(),
        DuckDbValue::UInt8(value) => value.to_string(),
        DuckDbValue::UInt16(value) => value.to_string(),
        DuckDbValue::UInt32(value) => value.to_string(),
        DuckDbValue::UInt64(value) => value.to_string(),
        DuckDbValue::UInt128(value) => value.to_string(),
        DuckDbValue::Float32(value) => value.to_string(),
        DuckDbValue::Float64(value) => value.to_string(),
        DuckDbValue::Decimal { value, scale, .. } => quack_decimal_text(&value.to_string(), *scale),
        DuckDbValue::Utf8(value) => ducklake_sql_string_literal(value),
        DuckDbValue::Type(value) => value.clone(),
        other => format!("{other:?}"),
    }
}

fn duckdb_value_type_name(value: &DuckDbValue) -> &'static str {
    match value {
        DuckDbValue::Null => "SQLNULL",
        DuckDbValue::Boolean(_) => "BOOLEAN",
        DuckDbValue::Int8(_) => "TINYINT",
        DuckDbValue::Int16(_) => "SMALLINT",
        DuckDbValue::Int32(_) => "INTEGER",
        DuckDbValue::Int64(_) => "BIGINT",
        DuckDbValue::Int128(_) => "HUGEINT",
        DuckDbValue::UInt8(_) => "UTINYINT",
        DuckDbValue::UInt16(_) => "USMALLINT",
        DuckDbValue::UInt32(_) => "UINTEGER",
        DuckDbValue::UInt64(_) => "UBIGINT",
        DuckDbValue::UInt128(_) => "UHUGEINT",
        DuckDbValue::Float32(_) => "FLOAT",
        DuckDbValue::Float64(_) => "DOUBLE",
        DuckDbValue::Utf8(_) => "VARCHAR",
        DuckDbValue::Blob(_) => "BLOB",
        DuckDbValue::DateDays(_) => "DATE",
        DuckDbValue::TimeMicros(_) => "TIME",
        DuckDbValue::TimeNanos(_) => "TIME_NS",
        DuckDbValue::TimeTzMicros(_) => "TIME_TZ",
        DuckDbValue::TimestampSeconds(_) => "TIMESTAMP_SEC",
        DuckDbValue::TimestampMillis(_) => "TIMESTAMP_MS",
        DuckDbValue::TimestampMicros(_) => "TIMESTAMP",
        DuckDbValue::TimestampNanos(_) => "TIMESTAMP_NS",
        DuckDbValue::TimestampTzMicros(_) => "TIMESTAMP_TZ",
        DuckDbValue::Decimal { .. } => "DECIMAL",
        DuckDbValue::Interval { .. } => "INTERVAL",
        DuckDbValue::Uuid(_) => "UUID",
        DuckDbValue::Type(_) => "TYPE",
        DuckDbValue::Struct(_) => "STRUCT",
        DuckDbValue::List(_) => "LIST",
        DuckDbValue::Map(_) => "MAP",
        DuckDbValue::Array(_) => "ARRAY",
    }
}

#[derive(Debug, Clone, PartialEq)]
pub(crate) struct DuckDbStructField {
    name: String,
    logical_type: DuckDbLogicalType,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct DuckDbListEntry {
    offset: usize,
    length: usize,
}

#[derive(Debug, Clone, PartialEq)]
pub(crate) enum DuckDbExtraTypeInfo {
    Generic {
        _alias: String,
    },
    Decimal {
        width: u8,
        scale: u8,
        _alias: String,
    },
    String {
        collation: String,
        _alias: String,
    },
    List {
        child_type: Box<DuckDbLogicalType>,
        _alias: String,
    },
    Struct {
        children: Vec<DuckDbStructField>,
        _alias: String,
    },
    Enum {
        values: Vec<String>,
        _alias: String,
    },
    Unbound {
        type_name: String,
        _alias: String,
    },
    AggregateState {
        function_name: String,
        return_type: Box<DuckDbLogicalType>,
        bound_argument_types: Vec<DuckDbLogicalType>,
        _alias: String,
    },
    Array {
        child_type: Box<DuckDbLogicalType>,
        size: u32,
        _alias: String,
    },
    Any {
        _target_type: Box<DuckDbLogicalType>,
        _cast_score: u64,
        _alias: String,
    },
    IntegerLiteral {
        constant_value: DuckDbValue,
        _alias: String,
    },
    Template {
        name: String,
        _alias: String,
    },
    Geo {
        crs_definition: String,
        _alias: String,
    },
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum DuckDbPhysicalType {
    Bool,
    Int8,
    Int16,
    Int32,
    Int64,
    Int128,
    UInt8,
    UInt16,
    UInt32,
    UInt64,
    UInt128,
    Float32,
    Float64,
    Interval,
    Varchar,
    Blob,
    Struct,
    List,
    Array,
}

#[derive(Debug, Clone, PartialEq)]
pub struct PrepareResponse {
    pub header: MessageHeader,
    pub result_types: Vec<String>,
    pub result_names: Vec<String>,
    pub needs_more_fetch: bool,
    pub results: Vec<DuckDbDataChunk>,
    pub result_uuid: i128,
}

#[derive(Debug, Clone, PartialEq)]
pub struct FetchResponse {
    pub header: MessageHeader,
    pub results: Vec<DuckDbDataChunk>,
    pub batch_index: Option<u64>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SuccessResponse {
    pub header: MessageHeader,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ErrorResponse {
    pub header: MessageHeader,
    pub message: String,
}

#[derive(Debug, Clone, PartialEq)]
pub enum QuackResponse {
    Connection(ConnectionResponse),
    Prepare(PrepareResponse),
    Fetch(FetchResponse),
    Success(SuccessResponse),
    Error(ErrorResponse),
}

impl QuackResponse {
    pub fn from_duckdb_binary(input: &[u8]) -> Result<Self> {
        let mut reader = DuckDbBinaryReader::new(input);
        let header = reader.read_header()?;
        let response = reader.read_response(header)?;
        reader.finish()?;
        Ok(response)
    }

    pub fn header(&self) -> &MessageHeader {
        match self {
            Self::Connection(response) => &response.header,
            Self::Prepare(response) => &response.header,
            Self::Fetch(response) => &response.header,
            Self::Success(response) => &response.header,
            Self::Error(response) => &response.header,
        }
    }

    pub fn message_type(&self) -> MessageType {
        self.header().message_type
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct QuackConnection {
    uri: QuackUri,
    connection_id: String,
}

impl QuackConnection {
    pub fn uri(&self) -> &QuackUri {
        &self.uri
    }

    pub fn from_response(uri: QuackUri, response: ConnectionResponse) -> Result<Self> {
        negotiate_quack_version(
            QUACK_VERSION,
            QUACK_VERSION,
            response.quack_version,
            &response.server_duckdb_version,
            &response.server_platform,
        )?;
        if response.header.connection_id.is_empty() {
            return Err(StorageError::Backend(
                "Quack server returned an empty connection id".into(),
            ));
        }
        Ok(Self {
            uri,
            connection_id: response.header.connection_id,
        })
    }

    pub fn connection_id(&self) -> &str {
        &self.connection_id
    }

    pub fn prepare_request(&self, sql_query: impl Into<String>) -> QuackRequest {
        QuackRequest::Prepare(PrepareRequest::new(self.connection_id.clone(), sql_query))
    }

    pub fn fetch_request(&self, result_uuid: i128) -> QuackRequest {
        QuackRequest::Fetch(FetchRequest::new(self.connection_id.clone(), result_uuid))
    }

    pub fn disconnect_request(&self) -> QuackRequest {
        QuackRequest::Disconnect(DisconnectRequest::new(self.connection_id.clone()))
    }
}

#[derive(Clone)]
pub struct QuackHttpClient {
    uri: QuackUri,
}

impl QuackHttpClient {
    pub fn new(uri: QuackUri) -> Self {
        Self { uri }
    }

    pub fn endpoint(&self) -> String {
        self.uri.quack_endpoint()
    }

    pub async fn send(&self, request: &QuackRequest) -> Result<QuackResponse> {
        let response = self.send_unvalidated(request).await?;
        request.validate_response(&response)?;
        Ok(response)
    }

    pub(crate) async fn send_unvalidated(&self, request: &QuackRequest) -> Result<QuackResponse> {
        let request_body = request.to_duckdb_binary()?;
        let response = reqwest::Client::new()
            .post(self.endpoint())
            .header(reqwest::header::CONTENT_TYPE, "application/octet-stream")
            .body(request_body)
            .send()
            .await
            .map_err(|err| StorageError::Backend(format!("failed to send Quack request: {err}")))?;

        let status = response.status();
        let response_body = response.bytes().await.map_err(|err| {
            StorageError::Backend(format!("failed to read Quack response body: {err}"))
        })?;
        if !status.is_success() {
            return Err(StorageError::Backend(format!(
                "Quack server returned HTTP status {status}"
            )));
        }

        let response = QuackResponse::from_duckdb_binary(&response_body)?;
        Ok(response)
    }
}

impl fmt::Debug for QuackHttpClient {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("QuackHttpClient")
            .field("uri", &self.uri)
            .finish_non_exhaustive()
    }
}

pub fn negotiate_quack_version(
    min_supported: u64,
    max_supported: u64,
    server_version: u64,
    server_duckdb_version: &str,
    server_platform: &str,
) -> Result<u64> {
    if min_supported > max_supported {
        return Err(StorageError::Backend(
            "invalid Quack client version range".into(),
        ));
    }
    if server_version < min_supported || server_version > max_supported {
        return Err(StorageError::Unsupported(format!(
            "Quack server protocol version {server_version} from DuckDB {server_duckdb_version} on {server_platform}; supported range is {min_supported}..={max_supported}"
        )));
    }
    Ok(server_version)
}

fn duckdb_map_entries_from_structs(
    values: &[DuckDbValue],
) -> Result<Vec<(DuckDbValue, DuckDbValue)>> {
    let mut entries = Vec::with_capacity(values.len());
    for value in values {
        let DuckDbValue::Struct(fields) = value else {
            return Err(StorageError::Backend(format!(
                "DuckDB MAP entry expected STRUCT child, got {value:?}"
            )));
        };
        let key = fields
            .iter()
            .find(|(name, _)| name == "key")
            .map(|(_, value)| value.clone())
            .ok_or_else(|| StorageError::Backend("DuckDB MAP entry missing key field".into()))?;
        let value = fields
            .iter()
            .find(|(name, _)| name == "value")
            .map(|(_, value)| value.clone())
            .ok_or_else(|| StorageError::Backend("DuckDB MAP entry missing value field".into()))?;
        entries.push((key, value));
    }
    Ok(entries)
}

fn duckdb_sequence_value(logical_type: &DuckDbLogicalType, raw: i64) -> Result<DuckDbValue> {
    let value = match logical_type.id {
        11 => DuckDbValue::Int8(i8::try_from(raw).map_err(|_| {
            StorageError::Backend(format!("DuckDB sequence value {raw} exceeds TINYINT"))
        })?),
        12 => DuckDbValue::Int16(i16::try_from(raw).map_err(|_| {
            StorageError::Backend(format!("DuckDB sequence value {raw} exceeds SMALLINT"))
        })?),
        13 => DuckDbValue::Int32(i32::try_from(raw).map_err(|_| {
            StorageError::Backend(format!("DuckDB sequence value {raw} exceeds INTEGER"))
        })?),
        14 => DuckDbValue::Int64(raw),
        15 => DuckDbValue::DateDays(i32::try_from(raw).map_err(|_| {
            StorageError::Backend(format!("DuckDB sequence value {raw} exceeds DATE"))
        })?),
        16 => DuckDbValue::TimeMicros(raw),
        17 => DuckDbValue::TimestampSeconds(raw),
        18 => DuckDbValue::TimestampMillis(raw),
        19 => DuckDbValue::TimestampMicros(raw),
        20 => DuckDbValue::TimestampNanos(raw),
        21 => {
            let (width, scale) = logical_type.decimal_width_scale().ok_or_else(|| {
                StorageError::Backend("DuckDB DECIMAL logical type is missing width/scale".into())
            })?;
            DuckDbValue::Decimal {
                value: i128::from(raw),
                width,
                scale,
            }
        }
        28 => DuckDbValue::UInt8(u8::try_from(raw).map_err(|_| {
            StorageError::Backend(format!("DuckDB sequence value {raw} exceeds UTINYINT"))
        })?),
        29 => DuckDbValue::UInt16(u16::try_from(raw).map_err(|_| {
            StorageError::Backend(format!("DuckDB sequence value {raw} exceeds USMALLINT"))
        })?),
        30 => DuckDbValue::UInt32(u32::try_from(raw).map_err(|_| {
            StorageError::Backend(format!("DuckDB sequence value {raw} exceeds UINTEGER"))
        })?),
        31 => DuckDbValue::UInt64(u64::try_from(raw).map_err(|_| {
            StorageError::Backend(format!("DuckDB sequence value {raw} exceeds UBIGINT"))
        })?),
        32 => DuckDbValue::TimestampTzMicros(raw),
        34 => DuckDbValue::TimeTzMicros(raw),
        35 => DuckDbValue::TimeNanos(raw),
        49 => DuckDbValue::UInt128(u128::try_from(raw).map_err(|_| {
            StorageError::Backend(format!("DuckDB sequence value {raw} exceeds UHUGEINT"))
        })?),
        50 => DuckDbValue::Int128(i128::from(raw)),
        other => {
            return Err(StorageError::Unsupported(format!(
                "Quack DuckDB SEQUENCE_VECTOR decoding for LogicalTypeId {other} is not supported"
            )));
        }
    };
    Ok(value)
}

pub fn validate_token(token: &str) -> Result<()> {
    if token.len() < 4 {
        return Err(StorageError::Backend(
            "Quack server token must be at least 4 characters long".into(),
        ));
    }
    Ok(())
}

fn duckdb_logical_type(type_id: u64) -> Result<DuckDbLogicalType> {
    let name = match type_id {
        0 => "INVALID",
        1 => "SQLNULL",
        2 => "UNKNOWN",
        3 => "ANY",
        4 => "UNBOUND",
        5 => "TEMPLATE",
        6 => "TYPE",
        10 => "BOOLEAN",
        11 => "TINYINT",
        12 => "SMALLINT",
        13 => "INTEGER",
        14 => "BIGINT",
        15 => "DATE",
        16 => "TIME",
        17 => "TIMESTAMP_SEC",
        18 => "TIMESTAMP_MS",
        19 => "TIMESTAMP",
        20 => "TIMESTAMP_NS",
        21 => "DECIMAL",
        22 => "FLOAT",
        23 => "DOUBLE",
        24 => "CHAR",
        25 => "VARCHAR",
        26 => "BLOB",
        27 => "INTERVAL",
        28 => "UTINYINT",
        29 => "USMALLINT",
        30 => "UINTEGER",
        31 => "UBIGINT",
        32 => "TIMESTAMP_TZ",
        34 => "TIME_TZ",
        35 => "TIME_NS",
        36 => "BIT",
        37 => "STRING_LITERAL",
        38 => "INTEGER_LITERAL",
        39 => "BIGNUM",
        49 => "UHUGEINT",
        50 => "HUGEINT",
        51 => "POINTER",
        53 => "VALIDITY",
        54 => "UUID",
        60 => "GEOMETRY",
        100 => "STRUCT",
        101 => "LIST",
        102 => "MAP",
        103 => "TABLE",
        104 => "ENUM",
        105 => "AGGREGATE_STATE",
        106 => "LAMBDA",
        107 => "UNION",
        108 => "ARRAY",
        109 => "VARIANT",
        other => {
            return Err(StorageError::Unsupported(format!(
                "unknown DuckDB LogicalTypeId {other}"
            )));
        }
    };
    Ok(DuckDbLogicalType {
        id: type_id,
        name,
        extra_info: None,
    })
}

fn duckdb_physical_type(logical_type: &DuckDbLogicalType) -> Result<DuckDbPhysicalType> {
    let physical_type = match logical_type.id {
        10 => DuckDbPhysicalType::Bool,
        11 => DuckDbPhysicalType::Int8,
        12 => DuckDbPhysicalType::Int16,
        13 | 15 => DuckDbPhysicalType::Int32,
        14 | 16 | 17 | 18 | 19 | 20 | 32 | 34 | 35 => DuckDbPhysicalType::Int64,
        21 => {
            let (width, scale) = logical_type.decimal_width_scale().ok_or_else(|| {
                StorageError::Backend("DuckDB DECIMAL logical type is missing width/scale".into())
            })?;
            if !(1..=38).contains(&width) || scale > width {
                return Err(StorageError::Backend(format!(
                    "invalid DuckDB DECIMAL width/scale: {width},{scale}"
                )));
            }
            if width <= 4 {
                DuckDbPhysicalType::Int16
            } else if width <= 9 {
                DuckDbPhysicalType::Int32
            } else if width <= 18 {
                DuckDbPhysicalType::Int64
            } else {
                DuckDbPhysicalType::Int128
            }
        }
        22 => DuckDbPhysicalType::Float32,
        23 => DuckDbPhysicalType::Float64,
        24 | 25 | 39 => DuckDbPhysicalType::Varchar,
        26 => DuckDbPhysicalType::Blob,
        27 => DuckDbPhysicalType::Interval,
        28 => DuckDbPhysicalType::UInt8,
        29 => DuckDbPhysicalType::UInt16,
        30 => DuckDbPhysicalType::UInt32,
        31 => DuckDbPhysicalType::UInt64,
        49 => DuckDbPhysicalType::UInt128,
        50 | 54 => DuckDbPhysicalType::Int128,
        100 | 107 | 109 => DuckDbPhysicalType::Struct,
        101 | 102 => DuckDbPhysicalType::List,
        104 => {
            let values = logical_type.enum_values()?;
            if values.len() <= usize::from(u8::MAX) {
                DuckDbPhysicalType::UInt8
            } else if values.len() <= usize::from(u16::MAX) {
                DuckDbPhysicalType::UInt16
            } else {
                DuckDbPhysicalType::UInt32
            }
        }
        108 => DuckDbPhysicalType::Array,
        other => {
            return Err(StorageError::Unsupported(format!(
                "Quack DuckDB DataChunk vector decoding for LogicalTypeId {other} is not supported"
            )));
        }
    };
    Ok(physical_type)
}

fn duckdb_physical_type_width(physical_type: DuckDbPhysicalType) -> Result<usize> {
    let width = match physical_type {
        DuckDbPhysicalType::Bool | DuckDbPhysicalType::Int8 | DuckDbPhysicalType::UInt8 => 1,
        DuckDbPhysicalType::Int16 | DuckDbPhysicalType::UInt16 => 2,
        DuckDbPhysicalType::Int32 | DuckDbPhysicalType::UInt32 | DuckDbPhysicalType::Float32 => 4,
        DuckDbPhysicalType::Int64 | DuckDbPhysicalType::UInt64 | DuckDbPhysicalType::Float64 => 8,
        DuckDbPhysicalType::Int128 | DuckDbPhysicalType::UInt128 | DuckDbPhysicalType::Interval => {
            16
        }
        DuckDbPhysicalType::Varchar
        | DuckDbPhysicalType::Blob
        | DuckDbPhysicalType::Struct
        | DuckDbPhysicalType::List
        | DuckDbPhysicalType::Array => {
            return Err(StorageError::Backend(
                "variable-width DuckDB physical type has no fixed byte width".into(),
            ));
        }
    };
    Ok(width)
}

fn duckdb_validity_mask_size(row_count: usize) -> usize {
    row_count.div_ceil(64) * 8
}

fn duckdb_validity_row_is_valid(validity: &[u8], row_idx: usize) -> bool {
    let byte_idx = row_idx / 8;
    let bit_idx = row_idx % 8;
    validity
        .get(byte_idx)
        .is_some_and(|byte| (byte & (1 << bit_idx)) != 0)
}

fn read_u16_le(bytes: &[u8], offset: usize) -> Result<u16> {
    Ok(u16::from_le_bytes(read_array(bytes, offset)?))
}

fn read_i16_le(bytes: &[u8], offset: usize) -> Result<i16> {
    Ok(i16::from_le_bytes(read_array(bytes, offset)?))
}

fn read_u32_le(bytes: &[u8], offset: usize) -> Result<u32> {
    Ok(u32::from_le_bytes(read_array(bytes, offset)?))
}

fn read_i32_le(bytes: &[u8], offset: usize) -> Result<i32> {
    Ok(i32::from_le_bytes(read_array(bytes, offset)?))
}

fn read_u64_le(bytes: &[u8], offset: usize) -> Result<u64> {
    Ok(u64::from_le_bytes(read_array(bytes, offset)?))
}

fn read_i64_le(bytes: &[u8], offset: usize) -> Result<i64> {
    Ok(i64::from_le_bytes(read_array(bytes, offset)?))
}

fn read_f32_le(bytes: &[u8], offset: usize) -> Result<f32> {
    Ok(f32::from_le_bytes(read_array(bytes, offset)?))
}

fn read_f64_le(bytes: &[u8], offset: usize) -> Result<f64> {
    Ok(f64::from_le_bytes(read_array(bytes, offset)?))
}

fn read_duckdb_i128(bytes: &[u8], offset: usize) -> Result<i128> {
    let lower = i128::from(read_u64_le(bytes, offset)?);
    let upper = i128::from(read_i64_le(bytes, offset + 8)?);
    Ok((upper << 64) | lower)
}

fn read_duckdb_u128(bytes: &[u8], offset: usize) -> Result<u128> {
    let lower = u128::from(read_u64_le(bytes, offset)?);
    let upper = u128::from(read_u64_le(bytes, offset + 8)?);
    Ok((upper << 64) | lower)
}

fn read_duckdb_uuid_bytes(bytes: &[u8], offset: usize) -> Result<[u8; 16]> {
    let lower = read_u64_le(bytes, offset)?;
    let upper = read_i64_le(bytes, offset + 8)? as u64;
    Ok(read_duckdb_uuid_bytes_from_parts(upper as i64, lower))
}

fn read_duckdb_uuid_bytes_from_parts(upper: i64, lower: u64) -> [u8; 16] {
    let upper = upper as u64;
    let high_bytes = upper ^ (1_u64 << 63);
    let mut out = [0_u8; 16];
    out[..8].copy_from_slice(&high_bytes.to_be_bytes());
    out[8..].copy_from_slice(&lower.to_be_bytes());
    out
}

fn read_array<const N: usize>(bytes: &[u8], offset: usize) -> Result<[u8; N]> {
    let end = offset
        .checked_add(N)
        .ok_or_else(|| StorageError::Backend("DuckDB byte slice offset overflow".into()))?;
    bytes
        .get(offset..end)
        .ok_or_else(|| StorageError::Backend("truncated DuckDB fixed-width value".into()))?
        .try_into()
        .map_err(|_| StorageError::Backend("invalid DuckDB fixed-width value".into()))
}

fn quack_integer_bool(idx: usize, value: i64) -> Result<bool> {
    match value {
        0 => Ok(false),
        1 => Ok(true),
        _ => Err(StorageError::Backend(format!(
            "Quack result column {idx} expected BOOLEAN-compatible integer 0 or 1, got {value}"
        ))),
    }
}

pub(crate) fn quack_parse_timestamp_micros(raw: &str) -> Option<i64> {
    let value = raw.trim().replace('T', " ");
    if value.len() < 19 {
        return None;
    }

    let year = value.get(0..4)?.parse::<i64>().ok()?;
    let month = value.get(5..7)?.parse::<i64>().ok()?;
    let day = value.get(8..10)?.parse::<i64>().ok()?;
    let hour = value.get(11..13)?.parse::<i64>().ok()?;
    let minute = value.get(14..16)?.parse::<i64>().ok()?;
    let second = value.get(17..19)?.parse::<i64>().ok()?;

    let mut rest = value.get(19..).unwrap_or("").trim();
    let mut micros = 0;
    if let Some(fraction) = rest.strip_prefix('.') {
        let end = fraction
            .find(|ch: char| !ch.is_ascii_digit())
            .unwrap_or(fraction.len());
        let digits = &fraction[..end];
        let padded = format!("{digits:0<6}");
        micros = padded.get(..6)?.parse::<i64>().ok()?;
        rest = fraction[end..].trim();
    }

    let offset_seconds = quack_parse_timezone_offset_seconds(rest).unwrap_or(0);
    let days = quack_days_from_civil(year, month, day);
    let local_seconds = days * 86_400 + hour * 3_600 + minute * 60 + second;
    Some((local_seconds - offset_seconds) * 1_000_000 + micros)
}

fn quack_parse_timezone_offset_seconds(raw: &str) -> Option<i64> {
    let value = raw.trim();
    if value.is_empty() || value.eq_ignore_ascii_case("z") {
        return Some(0);
    }

    let sign = match value.as_bytes().first()? {
        b'+' => 1,
        b'-' => -1,
        _ => return None,
    };
    let body = &value[1..];
    let hour = body.get(0..2)?.parse::<i64>().ok()?;
    let minute = if body.len() >= 5 && body.as_bytes().get(2) == Some(&b':') {
        body.get(3..5)?.parse::<i64>().ok()?
    } else if body.len() >= 4 {
        body.get(2..4)?.parse::<i64>().ok()?
    } else {
        0
    };
    Some(sign * (hour * 3_600 + minute * 60))
}

fn quack_days_from_civil(year: i64, month: i64, day: i64) -> i64 {
    let year = year - i64::from(month <= 2);
    let era = if year >= 0 { year } else { year - 399 } / 400;
    let year_of_era = year - era * 400;
    let month_prime = month + if month > 2 { -3 } else { 9 };
    let day_of_year = (153 * month_prime + 2) / 5 + day - 1;
    let day_of_era = year_of_era * 365 + year_of_era / 4 - year_of_era / 100 + day_of_year;
    era * 146_097 + day_of_era - 719_468
}

pub(crate) fn quack_decimal_text(unscaled: &str, scale: u8) -> String {
    if scale == 0 {
        return unscaled.to_string();
    }

    let (sign, digits) = unscaled
        .strip_prefix('-')
        .map(|digits| ("-", digits))
        .unwrap_or(("", unscaled));
    let scale = usize::from(scale);
    if digits.len() <= scale {
        let mut out = String::with_capacity(sign.len() + 2 + scale);
        out.push_str(sign);
        out.push_str("0.");
        for _ in 0..(scale - digits.len()) {
            out.push('0');
        }
        out.push_str(digits);
        return out;
    }

    let split_at = digits.len() - scale;
    format!("{sign}{}.{}", &digits[..split_at], &digits[split_at..])
}
