vertica_rs/conn/

async_conn.rs

use bytes::{Buf, BufMut, BytesMut};
use md5::{Digest, Md5};
use std::collections::HashMap;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpStream;
use rust_decimal::Decimal;
use rust_decimal::prelude::{FromPrimitive, ToPrimitive};

use crate::conn::{ConnectionConfig, ConnectionState};
use crate::error::error_response::ErrorResponse;
use crate::error::{Result, VerticaError};
use crate::query::{ColumnInfo, QueryResult};
use crate::types::vertica_types;
use crate::types::Value;
use crate::Row;
/// Async version of ConnectionStream
#[derive(Debug)]
pub enum AsyncConnectionStream {
    /// 表示使用普通的 TCP 连接流
    Plain(TcpStream),
    #[cfg(feature = "tls")]
    /// 表示使用 SSL/TLS 加密的 TCP 连接流
    Ssl(tokio_native_tls::TlsStream<TcpStream>),
}

impl AsyncConnectionStream {
    /// 异步读取数据到指定缓冲区。
    ///
    /// 此方法会尝试从连接流中异步读取数据到提供的缓冲区 `buf` 中。
    /// 返回成功读取的字节数。
    ///
    /// # 参数
    /// - `buf`: 用于存储读取数据的缓冲区。
    ///
    /// # 返回值
    /// 返回一个 `Result`，其中包含成功读取的字节数或错误信息。
    pub async fn read_async(&mut self, buf: &mut [u8]) -> Result<usize> {
        match self {
            AsyncConnectionStream::Plain(stream) => stream
                .read(buf)
                .await
                .map_err(|e| VerticaError::Connection(format!("Read error: {}", e))),
            #[cfg(feature = "tls")]
            AsyncConnectionStream::Ssl(stream) => stream
                .read(buf)
                .await
                .map_err(|e| VerticaError::Connection(format!("SSL read error: {}", e))),
        }
    }

    /// 异步将整个缓冲区的数据写入连接流。
    ///
    /// 此方法会尝试将提供的缓冲区 `buf` 中的所有数据异步写入连接流。
    /// 如果写入成功，返回 `Ok(())`；否则返回包含错误信息的 `Err`。
    ///
    /// # 参数
    /// - `buf`: 要写入连接流的数据缓冲区。
    ///
    /// # 返回值
    /// 返回一个 `Result`，表示写入操作是否成功。
    pub async fn write_all_async(&mut self, buf: &[u8]) -> Result<()> {
        match self {
            AsyncConnectionStream::Plain(stream) => stream
                .write_all(buf)
                .await
                .map_err(|e| VerticaError::Connection(format!("Write error: {}", e))),
            #[cfg(feature = "tls")]
            AsyncConnectionStream::Ssl(stream) => stream
                .write_all(buf)
                .await
                .map_err(|e| VerticaError::Connection(format!("SSL write error: {}", e))),
        }
    }
}

/// Async version of Connection following vertica-sql-go patterns
#[derive(Debug)]
pub struct AsyncConnection {
    stream: Option<AsyncConnectionStream>,
    config: ConnectionConfig,
    state: ConnectionState,
    parameters: HashMap<String, String>,
    backend_pid: Option<u32>,
    secret_key: Option<u32>,
}

impl AsyncConnection {
    /// Create a new async connection
    pub async fn new(config: ConnectionConfig) -> Result<Self> {
        let mut conn = Self {
            stream: None,
            config,
            state: ConnectionState::Disconnected,
            parameters: HashMap::new(),
            backend_pid: None,
            secret_key: None,
        };
        conn.connect().await?;
        Ok(conn)
    }

    /// Create a new async connection from URL
    pub async fn new_from_url(url: &str) -> Result<Self> {
        let config = ConnectionConfig::from_url(url)?;
        Self::new(config).await
    }

    /// Async connect to database - following vertica-sql-go's establishSocketConnection pattern
    async fn connect(&mut self) -> Result<()> {
        log::debug!(
            "[ASYNC_CONNECT] Starting connection process to {}:{} as {}",
            self.config.host,
            self.config.port,
            self.config.username
        );
        self.state = ConnectionState::Connecting;

        // Build host list including backup servers
        let mut hosts = vec![format!("{}:{}", self.config.host, self.config.port)];
        if let Some(ref backup_servers) = self.config.backup_servers {
            hosts.extend(backup_servers.clone());
        }

        let mut _last_error = None;

        for host_port in &hosts {
            log::debug!("[ASYNC_CONNECT] Trying host: {}", host_port);

            match self.connect_to_host(host_port).await {
                Ok(stream) => {
                    log::debug!("[ASYNC_CONNECT] Successfully connected to {}", host_port);
                    self.stream = Some(stream);
                    self.state = ConnectionState::Handshake;

                    return self.handshake().await.map_err(|e| {
                        log::debug!("[ASYNC_CONNECT] Handshake failed: {}", e);
                        e
                    });
                }
                Err(e) => {
                    log::debug!("[ASYNC_CONNECT] Failed to connect to {}: {}", host_port, e);
                    _last_error = Some(e);
                    continue;
                }
            }
        }

        let error_msg = format!(
            "Failed to connect to any host. Last error: {}",
            _last_error.unwrap_or_else(|| VerticaError::Connection("Unknown error".to_string()))
        );
        log::debug!("[ASYNC_CONNECT] {}", error_msg);
        Err(VerticaError::Connection(error_msg))
    }

    /// Async connect to specific host
    async fn connect_to_host(&self, host_port: &str) -> Result<AsyncConnectionStream> {
        let parts: Vec<&str> = host_port.split(':').collect();
        if parts.len() != 2 {
            return Err(VerticaError::Connection(format!(
                "Invalid host:port format: {}",
                host_port
            )));
        }

        let host = parts[0];
        let port: u16 = parts[1]
            .parse()
            .map_err(|e| VerticaError::Connection(format!("Invalid port number: {}", e)))?;

        // Resolve hostname to IP addresses
        let addrs = tokio::net::lookup_host(format!("{}:{}", host, port))
            .await
            .map_err(|e| {
                VerticaError::Connection(format!("Failed to resolve host {}: {}", host, e))
            })?;

        let mut _last_error = None;

        for addr in addrs {
            log::debug!("[ASYNC_CONNECT] Trying IP address: {}", addr);

            match tokio::time::timeout(self.config.connect_timeout, TcpStream::connect(addr)).await
            {
                Ok(Ok(stream)) => {
                    // Set socket options like vertica-sql-go
                    stream.set_nodelay(true).map_err(|e| {
                        VerticaError::Connection(format!("Failed to set TCP_NODELAY: {}", e))
                    })?;

                    return Ok(AsyncConnectionStream::Plain(stream));
                }
                Ok(Err(e)) => {
                    log::debug!("[ASYNC_CONNECT] Failed to connect to {}: {}", addr, e);
                    _last_error = Some(e);
                    continue;
                }
                Err(_) => {
                    log::debug!("[ASYNC_CONNECT] Connection timeout to {}", addr);
                    _last_error = Some(std::io::Error::new(
                        std::io::ErrorKind::TimedOut,
                        "Connection timeout",
                    ));
                    continue;
                }
            }
        }

        Err(VerticaError::Connection(format!(
            "Failed to connect to {}",
            host_port
        )))
    }

    /// Async handshake
    async fn handshake(&mut self) -> Result<()> {
        log::debug!("[ASYNC_HANDSHAKE] Starting handshake process");
        let mut buf = BytesMut::with_capacity(1024);

        let username = self.config.username.clone();
        let database = self.config.database.clone();

        buf.put_i32(0); // Length placeholder
        buf.put_i32(196608); // Protocol version 3.0

        self.write_string_async(&mut buf, "user").await?;
        self.write_string_async(&mut buf, &username).await?;

        self.write_string_async(&mut buf, "database").await?;
        self.write_string_async(&mut buf, &database).await?;

        self.write_string_async(&mut buf, "client_encoding").await?;
        self.write_string_async(&mut buf, "UTF8").await?;

        self.write_string_async(&mut buf, "").await?; // Terminator

        let len = buf.len() as i32;
        buf[0..4].copy_from_slice(&len.to_be_bytes());

        log::debug!("[ASYNC_HANDSHAKE] Sending startup message ({} bytes)", len);
        self.write_all_async(&buf).await.map_err(|e| {
            log::debug!("[ASYNC_HANDSHAKE] Failed to send startup message: {}", e);
            e
        })?;

        log::debug!("[ASYNC_HANDSHAKE] Startup message sent, starting authentication");
        self.authenticate().await?;
        log::debug!("[ASYNC_HANDSHAKE] Handshake completed successfully");
        Ok(())
    }

    /// Async authentication
    async fn authenticate(&mut self) -> Result<()> {
        log::debug!("[ASYNC_AUTH] Starting authentication process");
        self.state = ConnectionState::Authenticating;

        let mut message_count = 0;
        loop {
            message_count += 1;
            log::debug!("[ASYNC_AUTH] Waiting for message #{}", message_count);

            let mut msg = self.read_message().await.map_err(|e| {
                log::debug!(
                    "[ASYNC_AUTH] Failed to read message #{}: {}",
                    message_count,
                    e
                );
                e
            })?;

            log::debug!(
                "[ASYNC_AUTH] Received message type: {}",
                msg.message_type as char
            );

            match msg.message_type {
                b'R' => {
                    // Authentication request
                    let auth_type = msg.read_i32().map_err(|e| {
                        log::debug!("[ASYNC_AUTH] Failed to read auth type: {}", e);
                        e
                    })?;
                    log::debug!("[ASYNC_AUTH] Authentication request type: {}", auth_type);

                    match auth_type {
                        0 => {
                            // Authentication successful
                            log::debug!("[ASYNC_AUTH] Authentication successful");
                        }
                        3 => {
                            // Clear text password
                            log::debug!("[ASYNC_AUTH] Clear text password required");
                            self.send_password_response().await?;
                        }
                        5 => {
                            // MD5 password
                            log::debug!("[ASYNC_AUTH] MD5 password required");
                            let salt = msg.read_bytes(4)?;
                            log::debug!("[ASYNC_AUTH] Salt: {:02x?}", salt);
                            self.send_md5_password_response(&salt).await?;
                        }
                        _ => {
                            log::error!(
                                "[ASYNC_AUTH] Unsupported authentication type: {}",
                                auth_type
                            );
                            return Err(VerticaError::Authentication(format!(
                                "Unsupported authentication type: {}",
                                auth_type
                            )));
                        }
                    }
                }
                b'K' => {
                    let pid = msg.read_u32()?;
                    let key = msg.read_u32()?;
                    self.backend_pid = Some(pid);
                    self.secret_key = Some(key);
                    log::debug!("[ASYNC_AUTH] Backend PID: {}, Secret key: {}", pid, key);
                }
                b'S' => {
                    log::debug!(
                        "[ASYNC_AUTH] Processing parameter message, data length: {}",
                        msg.data.len()
                    );

                    if msg.data.is_empty() {
                        log::warn!("[ASYNC_AUTH] Empty parameter message, skipping");
                        continue;
                    }

                    let name = match msg.read_string() {
                        Ok(name) => name,
                        Err(e) => {
                            log::warn!("[ASYNC_AUTH] Failed to read parameter name: {}", e);
                            continue;
                        }
                    };

                    let value = match msg.read_string() {
                        Ok(value) => value,
                        Err(e) => {
                            log::warn!(
                                "[ASYNC_AUTH] Failed to read parameter value for '{}': {}",
                                name,
                                e
                            );
                            continue;
                        }
                    };

                    self.parameters.insert(name.clone(), value.clone());
                    log::debug!("[ASYNC_AUTH] Parameter: {} = {}", name, value);
                }
                b'Z' => {
                    self.state = ConnectionState::Ready;
                    break;
                }
                _ => {
                    return Err(VerticaError::Connection(format!(
                        "Unexpected message type: {}",
                        msg.message_type as char
                    )));
                }
            }
        }

        Ok(())
    }

    /// Async send password response
    async fn send_password_response(&mut self) -> Result<()> {
        log::debug!("[ASYNC_AUTH] Sending clear text password");
        let password = self
            .config
            .password
            .clone()
            .ok_or_else(|| VerticaError::Authentication("No password provided".to_string()))?;

        let mut buf = BytesMut::with_capacity(1024);
        buf.put_u8(b'p');
        buf.put_i32(0); // Length placeholder
        self.write_string_async(&mut buf, &password).await?;

        let len = buf.len() as i32;
        buf[1..5].copy_from_slice(&(len - 1).to_be_bytes());

        self.write_all_async(&buf).await
    }

    /// Async send MD5 password response
    async fn send_md5_password_response(&mut self, salt: &[u8]) -> Result<()> {
        log::debug!("[ASYNC_AUTH] Sending MD5 password response");

        let user = self.config.username.clone();
        let password = self
            .config
            .password
            .clone()
            .ok_or_else(|| VerticaError::Authentication("No password provided".to_string()))?;

        // Calculate MD5 hash: md5(md5(password + user) + salt)
        let mut hasher = Md5::new();
        hasher.update(password.as_bytes());
        hasher.update(user.as_bytes());
        let hash1 = hasher.finalize();

        let mut hasher = Md5::new();
        hasher.update(&hash1);
        hasher.update(salt);
        let hash2 = hasher.finalize();

        let md5_response = format!("md5{:x}", hash2);

        let mut buf = BytesMut::with_capacity(1024);
        buf.put_u8(b'p');
        buf.put_i32(0); // Length placeholder
        self.write_string_async(&mut buf, &md5_response).await?;

        let len = buf.len() as i32;
        buf[1..5].copy_from_slice(&(len - 1).to_be_bytes());

        self.write_all_async(&buf).await
    }

    /// Async read message
    async fn read_message(&mut self) -> Result<AsyncMessage> {
        let mut header = [0u8; 5];
        self.read_exact_async(&mut header).await?;

        let message_type = header[0];
        let len = i32::from_be_bytes([header[1], header[2], header[3], header[4]]) - 4;

        log::debug!(
            "[ASYNC_READ] Message type: {} ({}), length: {}",
            message_type as char,
            message_type,
            len
        );

        if len < 0 || len > 1_000_000 {
            return Err(VerticaError::Connection(format!(
                "Invalid message length: {}",
                len
            )));
        }

        let mut data = vec![0u8; len as usize];
        if len > 0 {
            self.read_exact_async(&mut data).await?;
        }

        log::debug!(
            "[ASYNC_READ] Successfully read message, data length: {}",
            data.len()
        );

        Ok(AsyncMessage {
            message_type,
            data: BytesMut::from(&data[..]),
        })
    }

    /// Async read exact bytes
    async fn read_exact_async(&mut self, buf: &mut [u8]) -> Result<()> {
        let mut read_index = 0;
        let total_len = buf.len();

        while read_index < total_len {
            let bytes_read = match &mut self.stream {
                Some(stream) => stream.read_async(&mut buf[read_index..]).await?,
                None => return Err(VerticaError::Connection("No active connection".to_string())),
            };

            if bytes_read == 0 {
                return Err(VerticaError::Connection(
                    "Connection closed by server".to_string(),
                ));
            }

            read_index += bytes_read;
            log::debug!(
                "[ASYNC_READ] Progress: {}/{} bytes ({} bytes this read)",
                read_index,
                total_len,
                bytes_read
            );
        }
        Ok(())
    }

    /// Async write all
    async fn write_all_async(&mut self, buf: &[u8]) -> Result<()> {
        match &mut self.stream {
            Some(stream) => stream.write_all_async(buf).await,
            None => Err(VerticaError::Connection("No active connection".to_string())),
        }
    }

    /// Async write string
    async fn write_string_async(&mut self, buf: &mut BytesMut, s: &str) -> Result<()> {
        buf.extend_from_slice(s.as_bytes());
        buf.put_u8(0); // null terminator
        Ok(())
    }

    /// Async simple query
    pub async fn simple_query(&mut self, query: &str) -> Result<QueryResult> {
        if self.state != ConnectionState::Ready {
            return Err(VerticaError::Connection("Connection not ready".to_string()));
        }

        let mut buf = BytesMut::with_capacity(1024);
        buf.put_u8(b'Q');
        buf.put_i32(0); // Length placeholder
        self.write_string_async(&mut buf, query).await?;

        let len = buf.len() as i32;
        buf[1..5].copy_from_slice(&(len - 1).to_be_bytes());

        self.write_all_async(&buf).await?;
        self.process_query_result().await
    }

    /// Async process query result
    async fn process_query_result(&mut self) -> Result<QueryResult> {
        let mut columns = Vec::new();
        let mut rows = Vec::new();

        loop {
            let mut msg = self.read_message().await?;

            match msg.message_type {
                b'T' => {
                    log::debug!("[ASYNC_READ] Processing row description message");
                    // Row description
                    let field_count = msg.read_i16()?;
                    for _ in 0..field_count {
                        let name = msg.read_string()?;
                        let _table_oid = msg.read_u32()?;
                        let _column_attr_number = msg.read_i16()?;
                        let data_type = msg.read_u32()?;
                        let _type_size = msg.read_i16()?;
                        let _type_modifier = msg.read_i32()?;
                        let _format_code = msg.read_i16()?;

                        columns.push(ColumnInfo {
                            name,
                            data_type,
                            nullable: true, // Default value
                            precision: None,
                            scale: None,
                        });
                    }
                }
                b'D' => {
                    log::debug!("[ASYNC_READ] Processing data row message");
                    // Data row
                    let column_count = msg.read_i16()?;
                    let mut row = Vec::new();

                    for i in 0..column_count {
                        let len = msg.read_i32()?;
                        if len == -1 {
                            row.push(Value::Null);
                        } else {
                            let mut data = vec![0u8; len as usize];
                            msg.read_exact(&mut data)?;

                            // 根据数据类型OID进行类型转换
                            let data_type = if (i as usize) < columns.len() {
                                columns[i as usize].data_type
                            } else {
                                0 // 默认类型
                            };

                            let value = Self::parse_binary_value(data_type, &data)?;
                            row.push(value);
                        }
                    }
                    rows.push(row);
                }
                b'C' => {
                    log::debug!("[ASYNC_READ] Processing command complete message");
                    // Command complete
                    let _tag = msg.read_string()?;
                    break;
                }
                b'Z' => {
                    log::debug!("[ASYNC_READ] Processing ready for query message");
                    // Ready for query
                    let _status = msg.read_u8()?;
                    break;
                }
                b'E' => {
                    log::debug!("[ASYNC_READ] Processing error response message");
                    // Error response - parse structured error
                    let error_response = ErrorResponse::parse_from_message(&msg.data)?;
                    log::error!("[ASYNC_READ] Server error: {}", error_response);
                    return Err(error_response.to_error());
                }
                b'S' => {
                    log::debug!("[ASYNC_READ] Processing notice response message");
                    let fields = msg.read_string()?;
                    log::info!("Notice: {}", fields);
                    continue;
                }
                b'N' => {
                    log::debug!("[ASYNC_READ] Processing notice response message");
                    let fields = msg.read_string()?;
                    log::info!("Notice: {}", fields);
                    continue;
                }
                b't' => {
                    // ParameterDescription - 参数描述
                    let param_count = msg.read_i16()?;
                    log::debug!("[PREPARE] ParameterDescription: {} parameters", param_count);

                    for i in 0..param_count {
                        let oid = msg.read_i32()?;
                        log::debug!("[PREPARE]   Parameter {}: OID {}", i, oid);
                    }
                }
                b'1' => {
                    log::debug!("[ASYNC_READ] Processing parse complete message");
                    // Parse complete - just continue
                    continue;
                }
                b'2' => {
                    log::debug!("[ASYNC_READ] Processing bind complete message");
                    // Bind complete - just continue
                    continue;
                }
                b's' => {
                    log::debug!("[ASYNC_READ] Processing portal suspended message");
                    // Portal suspended - end of results
                    continue;
                }
                _ => {
                    log::warn!(
                        "[ASYNC_READ] Unknown message type: {}",
                        msg.message_type as char
                    );
                    continue;
                }
            }
        }

        Ok(QueryResult::from_rows(rows, columns))
    }

    /// Get connection state
    pub fn state(&self) -> ConnectionState {
        self.state
    }

    /// Check if connection is ready
    pub fn is_ready(&self) -> bool {
        self.state == ConnectionState::Ready
    }

    /// Get server parameters
    pub fn parameters(&self) -> &HashMap<String, String> {
        &self.parameters
    }

    /// Parse binary data based on Vertica type OID
    fn parse_binary_value(data_type: u32, data: &[u8]) -> Result<Value> {
        use byteorder::{ByteOrder, LittleEndian, BigEndian};
        log::debug!(
            "[PARSE_BINARY_VALUE] data_type: {}, data_len: {}, raw_data: {:?}",
            data_type,
            data.len(),
            data
        );

        if data.is_empty() {
            log::debug!("[PARSE_BINARY_VALUE] Empty data, returning empty string");
            return Ok(Value::String(String::new()));
        }

        match data_type {
            vertica_types::BOOLEAN => {
                if data.len() >= 1 {
                    let val = Value::Boolean(data[0] != 0);
                    log::debug!("[PARSE_BOOLEAN] {:?}", val);
                    Ok(val)
                } else {
                    log::debug!("[PARSE_BOOLEAN] Empty data, defaulting to false");
                    Ok(Value::Boolean(false))
                }
            }
            vertica_types::INTEGER => {
                if data.len() >= 4 {
                    let val = Value::Int(LittleEndian::read_i32(data));
                    log::debug!("[PARSE_INTEGER] {:?}", val);
                    Ok(val)
                } else if data.len() == 2 {
                    let val = Value::Int(LittleEndian::read_i16(data) as i32);
                    log::debug!("[PARSE_INTEGER] 2-byte value: {:?}", val);
                    Ok(val)
                } else if data.len() == 1 {
                    let val = Value::Int(data[0] as i8 as i32);
                    log::debug!("[PARSE_INTEGER] 1-byte value: {:?}", val);
                    Ok(val)
                } else {
                    // 尝试解析字符串格式的整数
                    if let Ok(s) = String::from_utf8(data.to_vec()) {
                        if let Ok(num) = s.trim().parse::<i32>() {
                            log::debug!("[PARSE_INTEGER] String format: {:?}", num);
                            Ok(Value::Int(num))
                        } else if let Ok(num) = s.trim().parse::<i64>() {
                            // 如果数字太大，尝试i64后转换为i32（可能会截断）
                            log::debug!("[PARSE_INTEGER] Large string format: {} -> {}", num, num as i32);
                            Ok(Value::Int(num as i32))
                        } else {
                            log::debug!("[PARSE_INTEGER] Invalid string format, returning string: {}", s);
                            Ok(Value::String(s))
                        }
                    } else {
                        log::debug!("[PARSE_INTEGER] Unexpected len and invalid UTF-8, returning binary");
                        Ok(Value::Binary(data.to_vec()))
                    }
                }
            }
            vertica_types::BIGINT => {
                if data.len() >= 8 {
                    let val = Value::BigInt(LittleEndian::read_i64(data));
                    log::debug!("[PARSE_BIGINT] {:?}", val);
                    Ok(val)
                } else if data.len() == 4 {
                    let val = Value::BigInt(LittleEndian::read_i32(data) as i64);
                    log::debug!("[PARSE_BIGINT] 4-byte value: {:?}", val);
                    Ok(val)
                } else {
                    // 尝试解析字符串格式的整数
                    if let Ok(s) = String::from_utf8(data.to_vec()) {
                        if let Ok(num) = s.trim().parse::<i64>() {
                            log::debug!("[PARSE_BIGINT] String format: {:?}", num);
                            Ok(Value::BigInt(num))
                        } else {
                            log::debug!("[PARSE_BIGINT] Invalid string format, returning string: {}", s);
                            Ok(Value::String(s))
                        }
                    } else {
                        log::debug!("[PARSE_BIGINT] Unexpected len and invalid UTF-8, returning binary");
                        Ok(Value::Binary(data.to_vec()))
                    }
                }
            }
            vertica_types::FLOAT => {
                log::debug!(
                    "[PARSE_FLOAT] data_type: {}, data_len: {}, raw_bytes: {:?}",
                    data_type,
                    data.len(),
                    data
                );
                match data.len() {
                    8 => {
                        // FLOAT8 (DOUBLE PRECISION) - 8字节大端序
                        let val = BigEndian::read_f64(data);
                        log::debug!("[PARSE_FLOAT] FLOAT8: {}", val);
                        Ok(Value::Double(val))
                    },
                    4 => {
                        // FLOAT4 (REAL) - 4字节大端序
                        let val = BigEndian::read_f32(data);
                        log::debug!("[PARSE_FLOAT] FLOAT4: {}", val);
                        Ok(Value::Float(val))
                    },
                    _ => {
                        // 字符串格式：尝试解析数值字符串
                        let s = String::from_utf8_lossy(data);
                        log::debug!("[PARSE_FLOAT] String format: {}", s);
                        
                        let s = s.trim();
                        if let Ok(num) = s.parse::<f64>() {
                            log::debug!("[PARSE_FLOAT] Parsed as f64: {}", num);
                            Ok(Value::Double(num))
                        } else if let Ok(num) = s.parse::<f32>() {
                            log::debug!("[PARSE_FLOAT] Parsed as f32: {}", num);
                            Ok(Value::Float(num))
                        } else {
                            Ok(Value::String(s.to_string()))
                        }
                    }
                }
            }
            vertica_types::CHAR | vertica_types::VARCHAR | vertica_types::LONG_VARCHAR => {
                let val = String::from_utf8(data.to_vec())
                    .map(Value::String)
                    .unwrap_or_else(|_| Value::Binary(data.to_vec()));
                log::debug!("[PARSE_STRING] {:?}", val);
                Ok(val)
            }
            vertica_types::DATE => {
                if data.len() == 8 {
                    // 二进制格式：8字节的小端序整数
                    let days = LittleEndian::read_i64(data);
                    log::debug!(
                        "[PARSE_DATE] Binary format - days since 2000-01-01: {}",
                        days
                    );
                    Value::from_date_days(days as i32)
                } else {
                    // 字符串格式：尝试解析日期字符串
                    if let Ok(s) = String::from_utf8(data.to_vec()) {
                        log::debug!("[PARSE_DATE] String format: {}", s);
                        // 尝试多种日期格式
                        let formats = ["%Y-%m-%d", "%Y-%m-%d %H:%M:%S", "%Y/%m/%d", "%d-%m-%Y"];

                        let mut result = Value::String(s.clone());
                        for fmt in formats.iter() {
                            if let Ok(date) = chrono::NaiveDate::parse_from_str(&s.trim(), fmt) {
                                result = Value::Date(date);
                                break;
                            }
                        }

                        Ok(result)
                    } else {
                        log::debug!("[PARSE_DATE] Invalid UTF-8 string, returning binary");
                        Ok(Value::Binary(data.to_vec()))
                    }
                }
            }
            vertica_types::TIME => {
                if data.len() == 8 {
                    // 二进制格式：8字节的小端序整数
                    let micros = LittleEndian::read_i64(data);
                    log::debug!(
                        "[PARSE_TIME] Binary format - micros since midnight: {}",
                        micros
                    );
                    Value::from_time_micros(micros)
                } else {
                    // 字符串格式：尝试解析时间字符串
                    if let Ok(s) = String::from_utf8(data.to_vec()) {
                        log::debug!("[PARSE_TIME] String format: {}", s);
                        // 尝试多种时间格式
                        let formats = ["%H:%M:%S", "%Y-%m-%d %H:%M:%S", "%H:%M"];

                        let mut result = Value::String(s.clone());
                        for fmt in formats.iter() {
                            if let Ok(time) = chrono::NaiveTime::parse_from_str(&s.trim(), fmt) {
                                result = Value::Time(time);
                                break;
                            }
                        }

                        Ok(result)
                    } else {
                        log::debug!("[PARSE_TIME] Invalid UTF-8 string, returning binary");
                        Ok(Value::Binary(data.to_vec()))
                    }
                }
            }
            vertica_types::TIMETZ => {
                if data.len() == 8 {
                    // 二进制格式：8字节的小端序整数（微秒数）
                    let micros = LittleEndian::read_i64(data);
                    log::debug!(
                        "[PARSE_TIMETZ] Binary format - micros since midnight: {}",
                        micros
                    );
                    Value::from_time_micros(micros)
                } else {
                    // 字符串格式：尝试解析时间字符串（可能包含时区）
                    if let Ok(s) = String::from_utf8(data.to_vec()) {
                        log::debug!(
                            "[PARSE_TIMETZ] String format: {}",
                            s
                        );
                        
                        let s = s.trim();
                        
                        // 尝试解析带时区的时间格式，如"14:28:09.956353+08"
                        let formats = [
                            "%H:%M:%S%.f",      // 14:28:09.956353
                            "%H:%M:%S",         // 14:28:09
                        ];

                        // 提取时间部分（去掉时区偏移）
                        let time_part = if let Some(pos) = s.find(|c: char| c == '+' || c == '-') {
                            &s[..pos]
                        } else {
                            s
                        };

                        let mut result = Value::String(s.to_string()); // 默认返回原始字符串
                        for fmt in formats.iter() {
                            if let Ok(time) = chrono::NaiveTime::parse_from_str(time_part, fmt) {
                                result = Value::Time(time);
                                log::debug!(
                                    "[PARSE_TIMETZ] Successfully parsed as Time: {:?}",
                                    time
                                );
                                break;
                            }
                        }

                        Ok(result)
                    } else {
                        log::debug!(
                            "[PARSE_TIMETZ] Invalid UTF-8 string, returning binary"
                        );
                        Ok(Value::Binary(data.to_vec()))
                    }
                }
            }
            vertica_types::TIMESTAMPTZ => {
                if data.len() == 8 {
                    // 二进制格式：8字节的小端序整数
                    let micros = LittleEndian::read_i64(data);
                    log::debug!(
                        "[PARSE_TIMESTAMPTZ] Binary format - micros since 2000-01-01: {}",
                        micros
                    );
                    Value::from_timestamp_micros(micros)
                } else {
                    // 字符串格式：尝试解析时间戳字符串
                    if let Ok(s) = String::from_utf8(data.to_vec()) {
                        log::debug!(
                            "[PARSE_TIMESTAMPTZ] String format: {}",
                            s
                        );
                        // 尝试多种时间格式，包括带时区的格式
                        let formats = [
                            "%Y-%m-%d %H:%M:%S%.f%#z",
                            "%Y-%m-%d %H:%M:%S%#z", 
                            "%Y-%m-%d %H:%M:%S%.f",
                            "%Y-%m-%d %H:%M:%S",
                            "%Y-%m-%d %H:%M",
                            "%Y-%m-%d",
                        ];

                        let mut result = Value::String(s.clone());
                        for fmt in formats.iter() {
                            if let Ok(timestamp) =
                                chrono::NaiveDateTime::parse_from_str(&s.trim(), fmt)
                            {
                                result = Value::Timestamp(timestamp);
                                break;
                            }
                        }

                        // 如果还没解析成功，尝试解析为日期再转换为时间戳
                        if let Value::String(_) = result {
                            if let Ok(date) =
                                chrono::NaiveDate::parse_from_str(&s.trim(), "%Y-%m-%d")
                            {
                                result = Value::Timestamp(date.and_hms_opt(0, 0, 0).unwrap());
                            }
                        }

                        Ok(result)
                    } else {
                        log::debug!(
                            "[PARSE_TIMESTAMPTZ] Unexpected len and invalid UTF-8, returning binary"
                        );
                        Ok(Value::Binary(data.to_vec()))
                    }
                }
            }
            vertica_types::TIMESTAMP => {
                if data.len() == 8 {
                    // 二进制格式：8字节的小端序整数
                    let micros = LittleEndian::read_i64(data);
                    log::debug!(
                        "[PARSE_TIMESTAMP] Binary format - micros since 2000-01-01: {}",
                        micros
                    );
                    Value::from_timestamp_micros(micros)
                } else if data.len() == 19 {
                    // 字符串格式：19字节的日期时间字符串 "YYYY-MM-DD HH:MM:SS"
                    if let Ok(s) = String::from_utf8(data.to_vec()) {
                        log::debug!("[PARSE_TIMESTAMP] String format: {}", s);
                        Value::parse_timestamp(&s)
                    } else {
                        log::debug!("[PARSE_TIMESTAMP] Invalid UTF-8 string");
                        Ok(Value::Binary(data.to_vec()))
                    }
                } else {
                    // 其他长度，尝试作为字符串处理
                    if let Ok(s) = String::from_utf8(data.to_vec()) {
                        log::debug!(
                            "[PARSE_TIMESTAMP] Variable string format: {}",
                            s
                        );
                        // 尝试多种时间格式
                        let formats = [
                            "%Y-%m-%d %H:%M:%S",
                            "%Y-%m-%d %H:%M:%S%.f",
                            "%Y-%m-%d %H:%M",
                            "%Y-%m-%d",
                        ];

                        let mut result = Value::String(s.clone());
                        for fmt in formats.iter() {
                            if let Ok(timestamp) =
                                chrono::NaiveDateTime::parse_from_str(&s.trim(), fmt)
                            {
                                result = Value::Timestamp(timestamp);
                                break;
                            }
                        }

                        // 如果还没解析成功，尝试解析为日期再转换为时间戳
                        if let Value::String(_) = result {
                            if let Ok(date) =
                                chrono::NaiveDate::parse_from_str(&s.trim(), "%Y-%m-%d")
                            {
                                result = Value::Timestamp(date.and_hms_opt(0, 0, 0).unwrap());
                            }
                        }

                        Ok(result)
                    } else {
                        log::debug!(
                            "[PARSE_TIMESTAMP] Unexpected len and invalid UTF-8, returning binary"
                        );
                        Ok(Value::Binary(data.to_vec()))
                    }
                }
            }
            vertica_types::NUMERIC => {
                log::debug!(
                    "[PARSE_NUMERIC] type_oid: {}, data_len: {}, raw_bytes: {:?}",
                    data_type,
                    data.len(),
                    data
                );
                match data.len() {
                    8 => {
                        // NUMERIC as 8-byte float - 尝试解析为Decimal
                        let val = BigEndian::read_f64(data);
                        log::debug!("[PARSE_NUMERIC] NUMERIC8: {}", val);
                        if let Some(decimal) = Decimal::from_f64(val) {
                            Ok(Value::Decimal(decimal))
                        } else {
                            Ok(Value::Double(val))
                        }
                    },
                    4 => {
                        // NUMERIC as 4-byte float - 尝试解析为Decimal
                        let val = BigEndian::read_f32(data);
                        log::debug!("[PARSE_NUMERIC] NUMERIC4: {}", val);
                        if let Some(decimal) = Decimal::from_f32(val) {
                            Ok(Value::Decimal(decimal))
                        } else {
                            Ok(Value::Float(val))
                        }
                    },
                    16 => {
                        // NUMERIC as decimal128 - 尝试解析为Decimal
                        let s = String::from_utf8_lossy(data);
                        log::debug!("[PARSE_NUMERIC] NUMERIC16 string: {}", s);
                        if let Ok(decimal) = s.trim().parse::<Decimal>() {
                            Ok(Value::Decimal(decimal))
                        } else if let Ok(num) = s.trim().parse::<f64>() {
                            Ok(Value::Double(num))
                        } else {
                            Ok(Value::String(s.to_string()))
                        }
                    },
                    _ => {
                        // 字符串格式：优先尝试解析为Decimal
                        let s = String::from_utf8_lossy(data);
                        log::debug!("[PARSE_NUMERIC] String format: {}", s);
                        
                        let s = s.trim();
                        if let Ok(decimal) = s.parse::<Decimal>() {
                            log::debug!("[PARSE_NUMERIC] Parsed as Decimal: {}", decimal);
                            Ok(Value::Decimal(decimal))
                        } else if let Ok(num) = s.parse::<f64>() {
                            log::debug!("[PARSE_NUMERIC] Parsed as f64: {}", num);
                            Ok(Value::Double(num))
                        } else if let Ok(num) = s.parse::<f32>() {
                            log::debug!("[PARSE_NUMERIC] Parsed as f32: {}", num);
                            Ok(Value::Float(num))
                        } else if let Ok(num) = s.parse::<i64>() {
                            log::debug!("[PARSE_NUMERIC] Parsed as i64: {}", num);
                            Ok(Value::BigInt(num))
                        } else if let Ok(num) = s.parse::<i32>() {
                            log::debug!("[PARSE_NUMERIC] Parsed as i32: {}", num);
                            Ok(Value::Int(num))
                        } else {
                            Ok(Value::String(s.to_string()))
                        }
                    }
                }
            }
            vertica_types::UUID => {
                if data.len() >= 16 {
                    let val = Value::from_uuid_bytes(data.to_vec());
                    log::debug!("[PARSE_UUID] {:?}", val);
                    val
                } else {
                    log::debug!("[PARSE_UUID] Unexpected len, returning binary");
                    Ok(Value::Binary(data.to_vec()))
                }
            }
            vertica_types::VARBINARY | vertica_types::LONG_VARBINARY | vertica_types::BINARY => {
                let val = Value::Binary(data.to_vec());
                log::debug!("[PARSE_BINARY] {} bytes", data.len());
                Ok(val)
            }
            _ => {
                if let Ok(s) = String::from_utf8(data.to_vec()) {
                    log::debug!("[PARSE_UNKNOWN] UTF-8 string: {}", s);
                    Ok(Value::String(s))
                } else {
                    log::debug!("[PARSE_UNKNOWN] Binary: {} bytes", data.len());
                    Ok(Value::Binary(data.to_vec()))
                }
            }
        }
    }

    /// Prepare a SQL statement - 修复实现，更健壮地处理非标准响应
    pub async fn prepare(&mut self, stmt_id: &str, query: &str) -> Result<()> {
        if self.state != ConnectionState::Ready {
            return Err(VerticaError::Connection("Connection not ready".to_string()));
        }

        log::debug!("[PREPARE] Starting prepare for statement: {}", stmt_id);
        log::debug!("[PREPARE] SQL: {}", query);

        // 1. 构建并发送Parse消息
        let mut parse_buf = BytesMut::with_capacity(1024);
        parse_buf.put_u8(b'P'); // Parse消息类型
        parse_buf.put_i32(0); // 长度占位符

        // 写入语句名
        self.write_string_async(&mut parse_buf, stmt_id).await?;
        // 写入查询字符串
        self.write_string_async(&mut parse_buf, query).await?;
        // 写入参数格式代码数量 (0个)
        parse_buf.put_i16(0);

        let len = parse_buf.len() as i32;
        parse_buf[1..5].copy_from_slice(&(len - 1).to_be_bytes());

        log::debug!("[PREPARE] Sending Parse message ({} bytes)", len);
        self.write_all_async(&parse_buf).await?;

        // 2. 构建并发送Describe消息 - 获取语句元数据
        let mut describe_buf = BytesMut::with_capacity(64);
        describe_buf.put_u8(b'D'); // Describe消息类型
        describe_buf.put_i32(0); // 长度占位符
        describe_buf.put_u8(b'S'); // 描述语句 (S = statement)
        self.write_string_async(&mut describe_buf, stmt_id).await?;

        let describe_len = describe_buf.len() as i32;
        describe_buf[1..5].copy_from_slice(&(describe_len - 1).to_be_bytes());

        log::debug!(
            "[PREPARE] Sending Describe message ({} bytes)",
            describe_len
        );
        self.write_all_async(&describe_buf).await?;

        // 3. 构建并发送Sync消息 - 同步点
        let mut sync_buf = BytesMut::with_capacity(16);
        sync_buf.put_u8(b'S'); // Sync消息类型
        sync_buf.put_i32(4); // 固定长度4字节

        log::debug!("[PREPARE] Sending Sync message");
        self.write_all_async(&sync_buf).await?;

        // 4. 处理服务器响应 - 更健壮的实现
        let mut parse_complete_received = false;
        let  ready_for_query_received :bool;


        loop {
            let mut msg = self.read_message().await?;

            match msg.message_type {
                b'1' => {
                    // ParseComplete - Parse完成确认
                    log::debug!("[PREPARE] ParseComplete received");
                    parse_complete_received = true;
                }

                b't' => {
                    // ParameterDescription - 参数描述
                    let param_count = msg.read_i16()?;
                    log::debug!("[PREPARE] ParameterDescription: {} parameters", param_count);

                    for i in 0..param_count {
                        let oid = msg.read_i32()?;
                        log::debug!("[PREPARE]   Parameter {}: OID {}", i, oid);
                    }
                }

                b'T' => {
                    // RowDescription - 结果集描述
                    let field_count = msg.read_i16()?;
                    log::debug!("[PREPARE] RowDescription: {} columns", field_count);

                    let mut columns: Vec<ColumnInfo> = Vec::new();
                    for i in 0..field_count {
                        let name = msg.read_string()?;
                       let type_oid = msg.read_u32()?;
                        let column = ColumnInfo {
                            name: name.clone(),
                            data_type: type_oid,
                            nullable: true,
                            precision: None,
                            scale: None,
                        };
                        columns.push(column);

                        log::debug!(
                            "[PREPARE]   Column {}: name={}, type_oid={}",
                            i,
                            name,
                            type_oid
                        );
                    }
                }

                b'n' => {
                    // NoData - 无结果集 (如INSERT语句)
                    log::debug!("[PREPARE] NoData received (no result set)");
                }

                b'E' => {
                    // ErrorResponse - 错误响应
                    let error_response = ErrorResponse::parse_from_message(&msg.data)?;
                    log::error!("[PREPARE] ErrorResponse: {}", error_response);
                    return Err(error_response.to_error());
                }

                b'S' => {
                    // ParameterStatus - 服务器参数更新
                    let name = msg.read_string()?;
                    let value = msg.read_string()?;
                    self.parameters.insert(name.clone(), value.clone());
                    log::debug!("[PREPARE] ParameterStatus: {} = {}", name, value);
                }

                b'Z' => {
                    // ReadyForQuery - 事务状态
                    let status = msg.read_u8()?;
                    log::debug!("[PREPARE] ReadyForQuery: status={}", status as char);
                    ready_for_query_received = true;

                    // 收到ReadyForQuery表示消息序列结束
                    break;
                }

                b'C' => {
                    // CommandComplete - 命令完成
                    let tag = msg.read_string()?;
                    log::debug!("[PREPARE] CommandComplete: {}", tag);
                }

                b'D' => {
                    // DataRow - 数据行 (在prepare阶段不应该出现，但跳过它)
                    let col_count = msg.read_i16()?;
                    log::warn!(
                        "[PREPARE] Unexpected DataRow: {} columns - skipping",
                        col_count
                    );

                    // 跳过数据内容
                    for _ in 0..col_count {
                        let len = msg.read_i32()?;
                        if len > 0 {
                            let _data = msg.read_bytes(len as usize)?;
                        }
                    }
                }

                b'N' => {
                    // NoticeResponse - 通知消息
                    let notice = msg.read_string()?;
                    log::info!("[PREPARE] Notice: {}", notice);
                }

                _ => {
                    // 未知消息类型 - 跳过
                    log::warn!(
                        "[PREPARE] Unknown message type: {} ({}) - skipping",
                        msg.message_type as char,
                        msg.message_type
                    );
                    continue;
                }
            }
        }

        // 检查是否收到必要消息，但不要求ParseComplete必须出现
        if !ready_for_query_received {
            log::warn!("[PREPARE] Missing ReadyForQuery, prepare may have failed");
            return Err(VerticaError::Connection(
                "Incomplete prepare response - missing ReadyForQuery".to_string(),
            ));
        }

        // 如果没有收到ParseComplete但收到了ReadyForQuery，可能是服务器行为差异
        if !parse_complete_received {
            log::warn!(
                "[PREPARE] No ParseComplete received, but continuing as server indicated ready"
            );
        }

        log::debug!("[PREPARE] Prepare process completed");
        Ok(())
    }

    /// Execute prepared statement
    pub async fn execute_prepared(
        &mut self,
        statement_id: &str,
        params: &[Value],
    ) -> Result<QueryResult> {
        log::debug!(
            "[EXECUTE_PREPARED] Executing prepared statement: {} with {} parameters",
            statement_id,
            params.len()
        );

        // 发送 Bind 消息
        let mut bind_msg = BytesMut::new();
        bind_msg.put_u8(b'B'); // Bind
        bind_msg.put_i32(0); // 占位长度

        // Portal name (空字符串表示未命名portal)
        bind_msg.put_slice(b"");
        bind_msg.put_u8(0);

        // Statement name
        bind_msg.put_slice(statement_id.as_bytes());
        bind_msg.put_u8(0);

        // 参数格式代码数量 (0 = 所有参数使用默认格式)
        bind_msg.put_i16(0);

        // 参数值数量
        bind_msg.put_i16(params.len() as i16);

        // 参数值
        for param in params {
            if let Value::Null = *param {
                bind_msg.put_i32(-1); // NULL 值
            } else {
                // 由于 `to_bytes` 方法不存在，推测应该使用 `Value::from_bytes` 的逆操作，假设存在 `to_binary` 方法
                // 若实际使用时不存在该方法，需根据 `Value` 类型的具体实现调整
                let bytes = param.as_bytes()?;
                bind_msg.put_i32(bytes.len() as i32);
                bind_msg.extend_from_slice(&bytes);
            }
        }

        // 结果列格式 (0 = 所有列使用默认格式)
        bind_msg.put_i16(0);

        // 更新长度
        let len = bind_msg.len() - 1;
        bind_msg[1..5].copy_from_slice(&(len as i32).to_be_bytes());

        // 发送 Execute 消息
        let mut execute_msg = BytesMut::new();
        execute_msg.put_u8(b'E'); // Execute
        execute_msg.put_i32(9); // 长度: 4字节(长度字段) + 1字节(空portal名) + 4字节(最大行数)
        execute_msg.put_u8(0); // Portal name (空字符串，1字节)
        execute_msg.put_i32(0); // 最大行数 (0 = 获取所有行，4字节)

        // 更新长度
        let len = execute_msg.len() - 1;
        execute_msg[1..5].copy_from_slice(&(len as i32).to_be_bytes());

        // 发送 Sync 消息
        let mut sync_msg = BytesMut::new();
        sync_msg.put_u8(b'S'); // Sync
        sync_msg.put_i32(4); // Length

        // 合并消息并发送
        let mut full_msg = BytesMut::new();
        full_msg.extend_from_slice(&bind_msg);
        full_msg.extend_from_slice(&execute_msg);
        full_msg.extend_from_slice(&sync_msg);

        if let Some(ref mut stream) = self.stream {
            stream.write_all_async(&full_msg).await?;
        } else {
            return Err(VerticaError::Connection(
                "Connection not established".to_string(),
            ));
        }

        log::debug!(
            "[EXECUTE_PREPARED] Sent Bind/Execute/Sync messages for statement: {}",
            statement_id
        );

        // 处理响应
        let mut rows = Vec::new();
        let mut columns = Vec::new();
        let mut affected_rows = 0;
        let mut command_tag = String::new();
        let mut command_complete_received = false;

        loop {
            let  msg = self.read_message().await?;
            log::debug!(
                "[EXECUTE_PREPARED] Received message type: {} ({:x})",
                msg.message_type as char,
                msg.message_type
            );

            let mut msg_data = AsyncMessage {
                message_type: msg.message_type,
                data: msg.data.clone(),
            };

            match msg.message_type {
                b't' => {
                    // ParameterDescription - 参数描述
                    log::debug!("[EXECUTE_PREPARED] Unexpected ParameterDescription - skipping");
                }
                b'T' => {
                    // RowDescription - 结果集描述
                    let field_count = msg_data.read_i16()?;
                    log::debug!("[EXECUTE_PREPARED] RowDescription: {} columns", field_count);

                    columns.clear();
                    for i in 0..field_count {
                        let name = msg_data.read_string()?;
                       let type_oid = msg_data.read_u32()?;
                        let column = ColumnInfo {
                            name: name.clone(),
                            data_type: type_oid,
                            nullable: true,
                            precision: None,
                            scale: None,
                        };
                        columns.push(column);

                        log::debug!(
                            "[EXECUTE_PREPARED]   Column {}: name={}, type_oid={}",
                            i,
                            &name,
                            type_oid
                        );
                    }
                }
                b'D' => {
                    // DataRow - 数据行
                    let col_count = msg_data.read_i16()?;
                    let mut values = Vec::with_capacity(col_count as usize);

                    for _i in 0..col_count {
                        let len = msg_data.read_i32()?;
                        if len == -1 {
                            // NULL 值
                            values.push(Value::Null);
                        } else if len >= 0 {
                            let data = msg_data.read_bytes(len as usize)?;
                            let value = Value::from_bytes(data);
                            values.push(value);
                        } else {
                            return Err(VerticaError::Protocol(
                                "Invalid data length in DataRow".to_string(),
                            ));
                        }
                    }

                    let row = Row { data: values };
                    rows.push(row);
                }
                b'C' => {
                    // CommandComplete - 命令完成
                    command_tag = msg_data.read_string()?;
                    log::debug!("[EXECUTE_PREPARED] CommandComplete: {}", command_tag);

                    // 解析影响的行数
                    if let Some(pos) = command_tag.rfind(' ') {
                        if let Ok(rows) = command_tag[pos + 1..].parse::<u64>() {
                            affected_rows = rows;
                        }
                    }
                    command_complete_received = true;
                }
                b'E' => {
                    // ErrorResponse - 错误响应
                    let error_response = ErrorResponse::parse_from_message(&msg.data)?;
                    log::error!("[EXECUTE_PREPARED] ErrorResponse: {}", error_response);
                    return Err(error_response.to_error());
                }
                b'Z' => {
                    // ReadyForQuery - 事务状态
                    let status = msg_data.read_u8()?;
                    log::debug!(
                        "[EXECUTE_PREPARED] ReadyForQuery: status={}",
                        status as char
                    );
                    break;
                }
                b'S' => {
                    // ParameterStatus - 服务器参数更新
                    let name = msg_data.read_string()?;
                    let value = msg_data.read_string()?;
                    self.parameters.insert(name.clone(), value.clone());
                    log::debug!("[EXECUTE_PREPARED] ParameterStatus: {} = {}", name, value);
                }
                b'N' => {
                    // NoticeResponse - 通知消息
                    let notice = msg_data.read_string()?;
                    log::info!("[EXECUTE_PREPARED] Notice: {}", notice);
                }
                b'n' => {
                    // NoData - 无结果集
                    log::debug!("[EXECUTE_PREPARED] NoData received");
                }
                b'1' => {
                    log::debug!("[ASYNC_READ] Processing parse complete message");
                    // Parse complete - just continue
                    continue;
                }
                b'2' => {
                    log::debug!("[ASYNC_READ] Processing bind complete message");
                    // Bind complete - just continue
                    continue;
                }
                b's' => {
                    log::debug!("[ASYNC_READ] Processing portal suspended message");
                    // Portal suspended - end of results
                    break;
                }
                _ => {
                    log::warn!(
                        "[ASYNC_READ] Unknown message type: {}",
                        msg.message_type as char
                    );
                    continue;
                }
            }
        }

        log::debug!(
            "[EXECUTE_PREPARED] Execution completed. Rows: {}, Columns: {}, Affected: {}",
            rows.len(),
            columns.len(),
            affected_rows
        );

        // 构建结果
        let result = if columns.is_empty() && command_complete_received {
            // 非查询命令 (INSERT, UPDATE, DELETE等)
            QueryResult::from_command(affected_rows as u64, Some(command_tag), "".to_owned())
        } else if !columns.is_empty() {
            // 查询命令
            QueryResult::from_rows(rows.into_iter().map(|r| r.data).collect(), columns)
        } else {
            // 空结果
            QueryResult::empty()
        };

        Ok(result)
    }

    /// Close connection
    pub async fn close(&mut self) -> Result<()> {
        if let Some(mut stream) = self.stream.take() {
            stream.write_all_async(&[]).await?;
        }
        self.state = ConnectionState::Disconnected;
        Ok(())
    }
}

/// Async message structure
pub struct AsyncMessage {
    /// 消息类型，使用 u8 表示消息的类型标识
    pub message_type: u8,
    /// 存储消息数据的缓冲区
    pub data: BytesMut,
}
/// 为 `AsyncMessage` 实现消息读取方法，用于从消息数据中解析各种类型的值。
impl AsyncMessage {
    /// 从消息数据中读取一个 `u8` 类型的值。
    ///
    /// 如果消息数据为空，将返回一个包含错误信息的 `Err`。
    ///
    /// # 返回值
    /// - `Ok(u8)`: 成功读取的 `u8` 值。
    /// - `Err(VerticaError)`: 读取失败时的错误信息。
    pub fn read_u8(&mut self) -> Result<u8> {
        if self.data.is_empty() {
            return Err(VerticaError::Connection(
                "read_u8:Unexpected end of message".to_string(),
            ));
        }
        Ok(self.data.get_u8())
    }

    /// 从消息数据中读取一个 `i16` 类型的值。
    ///
    /// 如果消息数据长度不足 2 字节，将返回一个包含错误信息的 `Err`。
    ///
    /// # 返回值
    /// - `Ok(i16)`: 成功读取的 `i16` 值。
    /// - `Err(VerticaError)`: 读取失败时的错误信息。
    pub fn read_i16(&mut self) -> Result<i16> {
        if self.data.len() < 2 {
            return Err(VerticaError::Connection(
                "read_i16:Unexpected end of message".to_string(),
            ));
        }
        Ok(self.data.get_i16())
    }

    /// 从消息数据中读取一个 `i32` 类型的值。
    ///
    /// 原错误信息存在笔误，当前代码中错误信息为 "read_i16"，正确应为 "read_i32"。
    /// 如果消息数据长度不足 4 字节，将返回一个包含错误信息的 `Err`。
    ///
    /// # 返回值
    /// - `Ok(i32)`: 成功读取的 `i32` 值。
    /// - `Err(VerticaError)`: 读取失败时的错误信息。
    pub fn read_i32(&mut self) -> Result<i32> {
        if self.data.len() < 4 {
            return Err(VerticaError::Connection(
                "read_i32:Unexpected end of message".to_string(),
            ));
        }
        Ok(self.data.get_i32())
    }

    /// 从消息数据中读取一个 `u32` 类型的值。
    ///
    /// 如果消息数据长度不足 4 字节，将返回一个包含错误信息的 `Err`。
    ///
    /// # 返回值
    /// - `Ok(u32)`: 成功读取的 `u32` 值。
    /// - `Err(VerticaError)`: 读取失败时的错误信息。
    pub fn read_u32(&mut self) -> Result<u32> {
        if self.data.len() < 4 {
            return Err(VerticaError::Connection(
                "read_u32:Unexpected end of message".to_string(),
            ));
        }
        Ok(self.data.get_u32())
    }

    /// 从消息数据中读取一个以空字符 (`\0`) 结尾的字符串。
    ///
    /// 该方法会处理空消息、缺少空字符结尾、空字符位置越界等情况，
    /// 并使用 `from_utf8_lossy` 方法优雅地处理无效的 UTF-8 编码。
    ///
    /// # 返回值
    /// - `Ok(String)`: 成功读取的字符串。
    /// - `Err(VerticaError)`: 读取失败时的错误信息。
    pub fn read_string(&mut self) -> Result<String> {
        if self.data.is_empty() {
            return Err(VerticaError::Connection(
                "Empty message when reading string".to_string(),
            ));
        }

        let null_pos = self.data.iter().position(|&b| b == 0).ok_or_else(|| {
            VerticaError::Connection("Invalid string format: no null terminator found".to_string())
        })?;

        if null_pos >= self.data.len() {
            return Err(VerticaError::Connection(
                "Invalid string format: null terminator beyond buffer".to_string(),
            ));
        }

        let string_data = self.data.split_to(null_pos);

        // 跳过空字符终止符
        if !self.data.is_empty() {
            self.data.advance(1);
        }

        // 优雅地处理空字符串
        if string_data.is_empty() {
            return Ok(String::new());
        }

        // 使用 from_utf8_lossy 优雅地处理无效的 UTF-8 编码，类似于 Go 实现
        let string = String::from_utf8_lossy(&string_data[..]).to_string();

        Ok(string)
    }

    /// Read a tagged string (field type + string value) similar to Go's readTaggedString
    /// 读取一个带标签的字符串（字段类型 + 字符串值），类似于 Go 的 `readTaggedString` 方法。
    ///
    /// 首先读取一个 `u8` 类型的字段类型，若字段类型为 0，则表示字符串为空。
    /// 否则，继续读取一个以空字符结尾的字符串。
    ///
    /// # 返回值
    /// - `Ok((u8, String))`: 成功读取的字段类型和对应的字符串。
    /// - `Err(VerticaError)`: 读取失败时的错误信息。
    pub fn read_tagged_string(&mut self) -> Result<(u8, String)> {
        if self.data.is_empty() {
            return Err(VerticaError::Connection(
                "Empty message when reading tagged string".to_string(),
            ));
        }

        let field_type = self.data.get_u8();

        // 处理字段类型为 0 的情况，表示空字符串（结束标记）
        if field_type == 0 {
            return Ok((0, String::new()));
        }

        let value = self.read_string()?;
        Ok((field_type, value))
    }

    /// 从消息数据中读取指定长度的字节数据。
    ///
    /// 如果消息数据长度不足指定长度，将返回一个包含错误信息的 `Err`。
    ///
    /// # 参数
    /// - `len`: 需要读取的字节长度。
    ///
    /// # 返回值
    /// - `Ok(Vec<u8>)`: 成功读取的字节数据。
    /// - `Err(VerticaError)`: 读取失败时的错误信息。
    pub fn read_bytes(&mut self, len: usize) -> Result<Vec<u8>> {
        if self.data.len() < len {
            return Err(VerticaError::Connection(
                "read_bytes:Unexpected end of message".to_string(),
            ));
        }
        Ok(self.data.split_to(len).to_vec())
    }

    /// 从消息数据中读取指定缓冲区长度的字节数据，并填充到缓冲区中。
    ///
    /// 如果消息数据长度不足缓冲区长度，将返回一个包含错误信息的 `Err`。
    ///
    /// # 参数
    /// - `buf`: 用于存储读取数据的缓冲区。
    ///
    /// # 返回值
    /// - `Ok(())`: 读取成功。
    /// - `Err(VerticaError)`: 读取失败时的错误信息。
    pub fn read_exact(&mut self, buf: &mut [u8]) -> Result<()> {
        if self.data.len() < buf.len() {
            return Err(VerticaError::Connection(
                "read_exact:Unexpected end of message".to_string(),
            ));
        }
        buf.copy_from_slice(&self.data.split_to(buf.len()));
        Ok(())
    }
}