br_db/types/

pgsql.rs

use crate::config::Connection;
use crate::types::{DbMode, DbResult, Mode, Params, TableOptions};
use br_pgsql::connect::Connect;
use br_pgsql::pools::Pools;
use br_pgsql::{PoolConstraints, PoolOpts};
use json::{array, object, JsonValue};
use lazy_static::lazy_static;
use log::{error, info, warn};
use std::collections::HashMap;
use std::sync::Arc;
use std::sync::Mutex;
use std::thread;
use std::time::Duration;

/// PostgreSQL 标识符最大长度（NAMEDATALEN - 1）
const PG_MAX_IDENTIFIER_LENGTH: usize = 63;

/// 生成 PostgreSQL 索引名称
/// 如果名称超过 63 个字符，使用 MD5 哈希缩短
fn generate_index_name(base_name: &str) -> String {
    if base_name.len() <= PG_MAX_IDENTIFIER_LENGTH {
        base_name.to_string()
    } else {
        // 使用 MD5 哈希生成固定长度的名称
        let md5 = br_crypto::md5::encrypt_hex(base_name.as_bytes());
        // 保留表名前缀的一部分，然后加上哈希值
        // 格式：{table}_hash_{hash}，确保总长度不超过 63 个字符
        if let Some(underscore_pos) = base_name.find('_') {
            // 计算可用长度：63 - "_hash_" (6) - MD5 前 16 个字符 (16) = 最多 41 个字符用于前缀
            let max_prefix_len = PG_MAX_IDENTIFIER_LENGTH - 6 - 16;
            let prefix = &base_name[..underscore_pos.min(max_prefix_len)];
            format!("{}_hash_{}", prefix, &md5[..16])
        } else {
            // 如果找不到下划线，直接使用哈希值
            format!("idx_{}", &md5[..(PG_MAX_IDENTIFIER_LENGTH - 4)])
        }
    }
}

lazy_static! {
    static ref TR: Arc<Mutex<HashMap<String, Arc<Mutex<Connect>>>>> =
        Arc::new(Mutex::new(HashMap::new()));
    static ref TRANS: Arc<Mutex<HashMap<String, i32>>> = Arc::new(Mutex::new(HashMap::new()));
    static ref TABLE_FIELDS: Arc<Mutex<HashMap<String, JsonValue>>> =
        Arc::new(Mutex::new(HashMap::new()));
    /// 连接池健康监控统计
    static ref POOL_STATS: Arc<Mutex<PoolStats>> = Arc::new(Mutex::new(PoolStats::new()));
}

/// 连接池健康状态
#[derive(Debug, Clone)]
pub struct PoolHealthStatus {
    /// 是否健康
    pub is_healthy: bool,
    /// 总连接数
    pub total_connections: usize,
    /// 活跃连接数
    pub active_connections: usize,
    /// 连接池使用率 (0.0 - 1.0)
    pub utilization: f64,
    /// 空闲连接数
    pub idle_connections: usize,
}

/// 连接池统计信息
#[derive(Debug, Clone)]
struct PoolStats {
    /// 总连接数
    total_connections: usize,
    /// 活跃连接数
    active_connections: usize,
    /// 空闲连接数
    idle_connections: usize,
    /// 连接创建总数
    connections_created: u64,
    /// 连接错误总数
    connection_errors: u64,
    /// 最后检查时间
    last_check: std::time::Instant,
}

impl PoolStats {
    fn new() -> Self {
        Self {
            total_connections: 0,
            active_connections: 0,
            idle_connections: 0,
            connections_created: 0,
            connection_errors: 0,
            last_check: std::time::Instant::now(),
        }
    }

    /// 记录连接创建
    fn record_connection_created(&mut self) {
        self.connections_created += 1;
        self.total_connections += 1;
        self.last_check = std::time::Instant::now();
    }

    /// 记录连接错误
    fn record_connection_error(&mut self) {
        self.connection_errors += 1;
        self.last_check = std::time::Instant::now();
    }

    /// 增加活跃连接
    fn increment_active(&mut self) {
        self.active_connections += 1;
        self.last_check = std::time::Instant::now();
    }

    /// 减少活跃连接
    fn decrement_active(&mut self) {
        if self.active_connections > 0 {
            self.active_connections -= 1;
        }
        self.last_check = std::time::Instant::now();
    }

    /// 更新空闲连接数
    fn update_idle_count(&mut self, idle_count: usize) {
        self.idle_connections = idle_count;
        self.last_check = std::time::Instant::now();
    }

    /// 获取连接池使用率
    fn utilization_rate(&self) -> f64 {
        if self.total_connections == 0 {
            0.0
        } else {
            self.active_connections as f64 / self.total_connections as f64
        }
    }
}

/// RAII连接计数器，确保离开作用域时自动减少活跃连接计数
struct ConnectionGuard;

impl ConnectionGuard {
    fn new() -> Self {
        Self
    }
}

impl Drop for ConnectionGuard {
    fn drop(&mut self) {
        // 自动减少活跃连接计数
        let mut stats = POOL_STATS.lock().unwrap();
        stats.decrement_active();
    }
}

#[derive(Clone)]
pub struct Pgsql {
    /// 当前连接配置
    pub connection: Connection,
    /// 当前选中配置
    pub default: String,
    pub params: Params,
    pub client: Pools,
}

impl Pgsql {
    /// 连接数据库（参考 mysql 的实现方式）
    /// 使用 PoolConstraints 和 PoolOpts 配置连接池
    pub fn connect(connection: Connection, default: String) -> Result<Self, String> {
        let port = connection
            .hostport
            .parse::<i32>()
            .map_err(|e| format!("端口号解析失败: {} ({})", connection.hostport, e))?;

        let cp_connection = connection.clone();
        let config = object! {
            debug: cp_connection.debug,
            username: cp_connection.username,
            userpass: cp_connection.userpass,
            database: cp_connection.database,
            hostname: cp_connection.hostname,
            hostport: port,
            charset: cp_connection.charset.str(),
        };
        let mut pgsql = br_pgsql::Pgsql::new(&config)?;

        // 配置连接池约束（参考 mysql 的实现）
        // 智能计算最优连接池大小：CPU核心数 × 2 + 1
        let cpu_cores = num_cpus::get();
        let optimal_pool_size = (cpu_cores * 2) + 1;

        // 从配置中读取 pool_max，如果没有则使用智能计算的默认值
        // 但是限制在合理范围内，避免过度连接
        let calculated_max = config["pool_max"]
            .as_u32()
            .filter(|&p| p > 0 && p <= 1000)
            .unwrap_or(std::cmp::min(optimal_pool_size as u32, 50))
            as usize; // 最大不超过50，保证稳定性

        info!(
            "PostgreSQL 连接池配置 - CPU核心数: {}, 计算最优值: {}, 实际使用: {}",
            cpu_cores, optimal_pool_size, calculated_max
        );

        let max_pools = calculated_max;

        let constraints =
            PoolConstraints::new(0, max_pools).map_err(|e| format!("连接池约束配置失败: {}", e))?;

        // 配置连接池选项（参考 mysql 的实现）
        let pool_opts = PoolOpts::default()
            .with_constraints(constraints)
            .with_reset_connection(true)
            .with_connect_timeout(Duration::from_secs(5))
            .with_read_timeout(Duration::from_secs(15))
            .with_write_timeout(Duration::from_secs(20))
            .with_tcp_keepalive(Duration::from_secs(5));

        // 使用新的连接池配置方式（参考 mysql 的实现）
        let pools = pgsql.pools_with_opts(pool_opts)?;

        // 初始化连接池统计
        {
            let mut stats = POOL_STATS.lock().unwrap();
            stats.record_connection_created();
        }

        Ok(Self {
            connection,
            default: default.clone(),
            params: Params::default("pgsql"),
            client: pools,
        })
    }

    /// 连接池健康检查（修复版）
    pub async fn health_check(&mut self) -> DbResult<PoolHealthStatus> {
        let _stats = POOL_STATS.lock().unwrap().clone();
        
        let is_healthy = {
            match self.client.get_guard() {
                Ok(mut guard) => {
                    match guard.conn().query("SELECT 1") {
                        Ok(_) => {
                            // 成功获取连接并执行查询，说明健康
                            drop(guard); // 确保连接返回池中
                            true
                        },
                        Err(e) => {
                            error!("连接池健康检查查询失败: {}", e);
                            false
                        }
                    }
                },
                Err(e) => {
                    error!("无法从连接池获取连接进行健康检查: {}", e);
                    false
                }
            }
        };
        
        let mut current_stats = POOL_STATS.lock().unwrap();
        
        // 更新总连接数（如果之前没有正确设置）
        if current_stats.total_connections == 0 {
            // 根据连接池配置设置合理的总连接数
            let cpu_cores = num_cpus::get();
            let optimal_pool_size = (cpu_cores * 2) + 1;
            current_stats.total_connections = std::cmp::min(optimal_pool_size, 50);
        }
        
        // 计算空闲连接数
        current_stats.idle_connections = if current_stats.total_connections >= current_stats.active_connections {
            current_stats.total_connections - current_stats.active_connections
        } else {
            0
        };
        
        Ok(PoolHealthStatus {
            is_healthy,
            total_connections: current_stats.total_connections,
            active_connections: current_stats.active_connections,
            utilization: current_stats.utilization_rate(),
            idle_connections: current_stats.idle_connections,
        })
    }

    fn query(&mut self, sql: &str) -> (bool, JsonValue) {
        let thread_id = format!("{:?}", thread::current().id());
        let key = format!("{}{}", self.default, thread_id);

        // === 连接池监控 ===
        // 使用RAII模式确保连接计数正确减少
        let _connection_guard = ConnectionGuard::new();

        {
            let mut stats = POOL_STATS.lock().unwrap();
            stats.increment_active();

            // 检查连接池健康状态
            let utilization = stats.utilization_rate();
            if utilization > 0.8 {
                warn!(
                    "PostgreSQL 连接池使用率过高: {:.1}%, 活跃连接: {}, 总连接: {}",
                    utilization * 100.0,
                    stats.active_connections,
                    stats.total_connections
                );
            }
        }

        // === 事务环境 ===
        // 修复：在单次锁持有期间同时检查 TRANS 和 TR，避免检查-使用时间窗口问题
        let db_opt = {
            let trans_map = TRANS.lock().unwrap_or_else(|e| e.into_inner());
            if trans_map.get(&*thread_id).is_some() {
                // 在事务环境中，同时获取 TR 锁查找连接
                let tr_map = TR.lock().unwrap_or_else(|e| e.into_inner());
                tr_map.get(&key).cloned()
            } else {
                None
            }
        };

        if let Some(db) = db_opt {
            // 事务环境：使用事务连接
            let mut t = db.lock().unwrap();
            match t.query(sql) {
                Ok(e) => {
                    // 查询成功（已移除正常日志）
                    (true, e.rows)
                }
                Err(e) => {
                    error!(
                        "事务查询失败 - 线程ID: {} | 数据库配置: {} | 数据库名: {} | SQL语句: [{}] | 错误详情: {}",
                        thread_id,
                        self.default,
                        self.connection.database,
                        sql,
                        e
                    );
                    (false, JsonValue::from(e.to_string()))
                }
            }
        } else {
            // 非事务环境下使用 ConnectionGuard
            let mut guard = match self.client.get_guard() {
                Ok(g) => g,
                Err(e) => {
                    error!(
                        "获取数据库连接失败 - 线程ID: {} | 数据库配置: {} | 数据库名: {} | SQL语句: [{}] | 错误详情: {}",
                        thread_id,
                        self.default,
                        self.connection.database,
                        sql,
                        e
                    );
                    return (false, JsonValue::from(e.to_string()));
                }
            };

            println!("DEBUG: About to execute query SQL: {}", sql);
            let res = guard.conn().query(sql);
            match res {
                Ok(e) => {
                    // 查询成功
                    println!("DEBUG: Query SQL executed successfully, returned {} rows", e.rows.len());
                    (true, e.rows)
                }
                Err(e) => {
                    println!("DEBUG: Query SQL failed with error: {}", e);
                    error!(
                        "非事务查询失败 - 线程ID: {} | 数据库配置: {} | 数据库名: {} | SQL语句: [{}] | 错误详情: {}",
                        thread_id,
                        self.default,
                        self.connection.database,
                        sql,
                        e
                    );
                    (false, JsonValue::from(e.to_string()))
                }
            }
            // guard 离开作用域时自动归还连接
        }
    }
    fn execute(&mut self, sql: &str) -> (bool, JsonValue) {
        let thread_id = format!("{:?}", thread::current().id());
        let key = format!("{}{}", self.default, thread_id);

        // === 事务环境 ===
        // 修复：在单次锁持有期间同时检查 TRANS 和 TR，避免检查-使用时间窗口问题
        let db_opt = {
            let trans_map = TRANS.lock().unwrap_or_else(|e| e.into_inner());
            if trans_map.get(&*thread_id).is_some() {
                // 在事务环境中，同时获取 TR 锁查找连接
                let tr_map = TR.lock().unwrap_or_else(|e| e.into_inner());
                tr_map.get(&key).cloned()
            } else {
                None
            }
        };

        if let Some(db) = db_opt {
            // 事务环境：使用事务连接
            let mut t = db.lock().unwrap();
            match t.execute(sql) {
                Ok(e) => {
                    // 提交成功（已移除正常日志）
                    if sql.contains("INSERT") {
                        (true, e.rows)
                    } else {
                        (true, e.affect_count.into())
                    }
                }
                Err(e) => {
                    let operation_type = if sql.trim_start().to_uppercase().starts_with("INSERT") {
                        "插入"
                    } else if sql.trim_start().to_uppercase().starts_with("UPDATE") {
                        "更新"
                    } else if sql.trim_start().to_uppercase().starts_with("DELETE") {
                        "删除"
                    } else {
                        "执行"
                    };
                    error!(
                        "事务{}失败 - 线程ID: {} | 数据库配置: {} | 数据库名: {} | 表名: {} | SQL语句: [{}] | 错误详情: {}",
                        operation_type,
                        thread_id,
                        self.default,
                        self.connection.database,
                        self.params.table,
                        sql,
                        e
                    );
                    (false, JsonValue::from(e.to_string()))
                }
            }
        } else {
            // 非事务环境下使用 ConnectionGuard
            let mut guard = match self.client.get_guard() {
                Ok(g) => g,
                Err(e) => {
                    error!(
                        "获取数据库连接失败 - 线程ID: {} | 数据库配置: {} | 数据库名: {} | SQL语句: [{}] | 错误详情: {}",
                        thread_id,
                        self.default,
                        self.connection.database,
                        sql,
                        e
                    );
                    return (false, JsonValue::from(e.to_string()));
                }
            };

            println!("DEBUG: About to execute SQL: {}", sql);
            let res = guard.conn().execute(sql);
            match res {
                Ok(e) => {
                    // 提交成功
                    println!("DEBUG: SQL executed successfully, affected rows: {}", e.affect_count);
                    if sql.contains("INSERT") {
                        (true, e.rows)
                    } else {
                        (true, e.affect_count.into())
                    }
                }
                Err(e) => {
                    println!("DEBUG: SQL failed with error: {}", e);
                    let operation_type = if sql.trim_start().to_uppercase().starts_with("INSERT") {
                        "插入"
                    } else if sql.trim_start().to_uppercase().starts_with("UPDATE") {
                        "更新"
                    } else if sql.trim_start().to_uppercase().starts_with("DELETE") {
                        "删除"
                    } else {
                        "执行"
                    };
                    error!(
                        "非事务{}失败 - 线程ID: {} | 数据库配置: {} | 数据库名: {} | 表名: {} | SQL语句: [{}] | 错误详情: {}",
                        operation_type,
                        thread_id,
                        self.default,
                        self.connection.database,
                        self.params.table,
                        sql,
                        e
                    );
                    (false, JsonValue::from(e.to_string()))
                }
            }
            // guard 离开作用域时自动归还连接
        }
    }
}

impl DbMode for Pgsql {
    fn database_tables(&mut self) -> JsonValue {
        let sql = "SELECT table_name FROM information_schema.tables 
             WHERE table_schema = 'public' AND table_type = 'BASE TABLE'"
            .to_string();
        match self.sql(sql.as_str()) {
            Ok(e) => {
                let mut list = vec![];
                for item in e.members() {
                    if let Some(value) = item["table_name"].as_str() {
                        list.push(JsonValue::from(value));
                    }
                }
                list.into()
            }
            Err(_) => {
                array![]
            }
        }
    }

    fn database_create(&mut self, name: &str) -> bool {
        // 先检查数据库是否已存在
        let check_sql = format!("SELECT 1 FROM pg_database WHERE datname = '{}'", name);
        let (exists_state, exists_data) = self.query(check_sql.as_str());

        if exists_state && !exists_data.is_empty() && exists_data.members().count() > 0 {
            // 数据库已存在，返回 true
            return true;
        }

        // 转义数据库名称（PostgreSQL 使用双引号）
        let db_name = format!("\"{}\"", name);

        // 构建 CREATE DATABASE SQL
        let mut sql = format!("CREATE DATABASE {}", db_name);

        // 添加字符集配置（PostgreSQL 使用 ENCODING）
        if !self.connection.charset.str().is_empty() {
            let charset_str = self.connection.charset.str();
            let encoding = match charset_str.as_str() {
                "utf8" | "utf8mb4" => "UTF8",
                "latin1" => "LATIN1",
                _ => "UTF8",
            };
            sql = format!("{} ENCODING '{}'", sql, encoding);
        }

        let (state, data) = self.execute(sql.as_str());
        match state {
            true => true, // PostgreSQL 的 CREATE DATABASE 成功时返回空结果
            false => {
                error!("创建数据库失败: {data:?}");
                false
            }
        }
    }

    fn database_update(&mut self, name: &str, options: JsonValue) -> bool {
        let db_name = format!("\"{}\"", name);
        let mut sql_parts = vec![];

        // 更新字符集
        if options.has_key("encoding") {
            let encoding = options["encoding"].as_str().unwrap_or("UTF8");
            sql_parts.push(format!("ENCODING '{}'", encoding));
        }

        // 更新所有者
        if options.has_key("owner") {
            let owner = options["owner"].as_str().unwrap_or("");
            sql_parts.push(format!("OWNER = \"{}\"", owner));
        }

        if sql_parts.is_empty() {
            return true; // 没有需要更新的内容
        }

        let sql = format!("ALTER DATABASE {} {}", db_name, sql_parts.join(" "));
        let (state, data) = self.execute(sql.as_str());

        match state {
            true => true,
            false => {
                error!("更新数据库失败: {data:?}");
                false
            }
        }
    }
}

impl Mode for Pgsql {
    fn transaction(&mut self) -> bool {
        let thread_id = format!("{:?}", thread::current().id());

        // 修复：在单次锁持有期间完成嵌套计数检查和更新，保证原子性
        let is_nested = {
            let mut trans_map = TRANS.lock().unwrap_or_else(|e| e.into_inner());
            if let Some(count) = trans_map.get_mut(&*thread_id) {
                // 嵌套事务：增加计数（在同一把锁内完成）
                *count += 1;
                true
            } else {
                // 新事务：初始化计数
                trans_map.insert(thread_id.clone(), 1);
                false
            }
        };

        if is_nested {
            return true;
        }

        let key = format!("{}{}", self.default, thread_id);

        // 获取连接并克隆（事务需要长期持有）
        let mut guard = match self.client.get_guard() {
            Ok(g) => g,
            Err(e) => {
                error!("获取事务连接失败: {e}");
                // 清理已插入的 TRANS 记录
                TRANS
                    .lock()
                    .unwrap_or_else(|e| e.into_inner())
                    .remove(&*thread_id);
                return false;
            }
        };

        let conn = guard.conn().clone();
        drop(guard);

        TR.lock()
            .unwrap_or_else(|e| e.into_inner())
            .insert(key.clone(), Arc::new(Mutex::new(conn)));

        let sql = "START TRANSACTION;".to_string();
        let (state, _) = self.execute(sql.as_str());
        match state {
            true => {
                let sql = "SET TRANSACTION ISOLATION LEVEL SERIALIZABLE;".to_string();
                let (state, _) = self.execute(sql.as_str());
                match state {
                    true => state,
                    false => {
                        // 清理资源：先 TRANS 后 TR（保持一致的锁顺序）
                        TRANS
                            .lock()
                            .unwrap_or_else(|e| e.into_inner())
                            .remove(&*thread_id);
                        TR.lock().unwrap_or_else(|e| e.into_inner()).remove(&key);
                        state
                    }
                }
            }
            false => {
                // 清理资源：先 TRANS 后 TR（保持一致的锁顺序）
                TRANS
                    .lock()
                    .unwrap_or_else(|e| e.into_inner())
                    .remove(&*thread_id);
                TR.lock().unwrap_or_else(|e| e.into_inner()).remove(&key);
                state
            }
        }
    }

    fn commit(&mut self) -> bool {
        let thread_id = format!("{:?}", thread::current().id());
        let key = format!("{}{}", self.default, thread_id);

        // 修复：在单次锁持有期间完成嵌套计数检查和更新，保证原子性
        let should_commit = {
            let mut trans_map = TRANS.lock().unwrap_or_else(|e| e.into_inner());
            if let Some(count) = trans_map.get_mut(&*thread_id) {
                if *count > 1 {
                    // 嵌套事务：减少计数（在同一把锁内完成）
                    *count -= 1;
                    false // 不需要执行 COMMIT
                } else {
                    // 最外层事务：标记为需要执行 COMMIT
                    true
                }
            } else {
                // 事务不存在，可能是已经被清理或从未创建
                error!("提交事务失败: 线程ID: {thread_id} 事务不存在");
                return false;
            }
        };

        if !should_commit {
            return true;
        }

        // 最外层事务：执行提交
        let sql = "COMMIT".to_string();
        let (state, data) = self.execute(sql.as_str());

        // 修复：仅在成功时清理资源，失败时保留资源以便重试或回滚
        if state {
            // 清理资源：先 TRANS 后 TR（保持一致的锁顺序）
            TRANS
                .lock()
                .unwrap_or_else(|e| e.into_inner())
                .remove(&thread_id);
            TR.lock().unwrap_or_else(|e| e.into_inner()).remove(&key);
        } else {
            let error_msg = data
                .as_str()
                .map(|s| s.to_string())
                .unwrap_or_else(|| data.to_string());
            error!(
                "提交事务失败 - 线程ID: {} | 数据库配置: {} | 数据库名: {} | SQL: {} | 错误详情: {}",
                thread_id,
                self.default,
                self.connection.database,
                sql,
                error_msg
            );
            // 注意：失败时不清理资源，保留 TRANS 和 TR 以便后续重试或回滚
        }
        state
    }

    fn rollback(&mut self) -> bool {
        let thread_id = format!("{:?}", thread::current().id());
        let key = format!("{}{}", self.default, thread_id);

        // 修复：在单次锁持有期间完成嵌套计数检查和更新，保证原子性
        let should_rollback = {
            let mut trans_map = TRANS.lock().unwrap_or_else(|e| e.into_inner());
            if let Some(count) = trans_map.get_mut(&*thread_id) {
                if *count > 1 {
                    // 嵌套事务：减少计数（在同一把锁内完成）
                    *count -= 1;
                    false // 不需要执行 ROLLBACK
                } else {
                    // 最外层事务：标记为需要执行 ROLLBACK
                    true
                }
            } else {
                // 事务不存在，可能是已经被清理或从未创建
                error!("回滚事务失败: 线程ID: {thread_id} 事务不存在");
                return false;
            }
        };

        if !should_rollback {
            return true;
        }

        // 最外层事务：执行回滚
        let sql = "ROLLBACK".to_string();
        let (state, data) = self.execute(sql.as_str());

        // 修复：无论成功与否都清理资源（回滚失败时也需要清理，避免资源泄漏）
        TRANS
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .remove(&thread_id);
        TR.lock().unwrap_or_else(|e| e.into_inner()).remove(&key);

        if !state {
            let error_msg = data
                .as_str()
                .map(|s| s.to_string())
                .unwrap_or_else(|| data.to_string());
            error!(
                "回滚事务失败 - 线程ID: {} | 数据库配置: {} | 数据库名: {} | SQL: {} | 错误详情: {}",
                thread_id,
                self.default,
                self.connection.database,
                sql,
                error_msg
            );
        }
        state
    }

    fn sql(&mut self, sql: &str) -> Result<JsonValue, String> {
        let (state, data) = self.query(sql);
        match state {
            true => Ok(data),
            false => Err(data.to_string()),
        }
    }

    fn sql_execute(&mut self, sql: &str) -> Result<JsonValue, String> {
        let (state, data) = self.execute(sql);
        match state {
            true => Ok(data),
            false => Err(data.to_string()),
        }
    }

    fn table_create(&mut self, options: TableOptions) -> JsonValue {
        println!("!!! TABLE_CREATE FUNCTION CALLED with table: {}", options.table_name);
        
        // 清除缓存
        let cache_key = format!("{}{}", self.default, options.table_name);
        if TABLE_FIELDS.lock().unwrap().get(&cache_key).is_some() {
            TABLE_FIELDS.lock().unwrap().remove(&cache_key);
        }

        let mut sql = String::new();
        let mut unique = String::new();
        let mut index = String::new();

        // 处理唯一索引
        let mut unique_fields = String::new();
        for item in options.table_unique.iter() {
            if unique_fields.is_empty() {
                unique_fields = format!("\"{}\"", item);
            } else {
                unique_fields = format!("{}, \"{}\"", unique_fields, item);
            }
        }
        if !unique_fields.is_empty() {
            let mut unique_name = String::new();
            for item in options.table_unique.iter() {
                if unique_name.is_empty() {
                    unique_name = format!("{}_unique_{}", options.table_name, item);
                } else {
                    unique_name = format!("{}_{}", unique_name, item);
                }
            }
            let md5 = br_crypto::md5::encrypt_hex(unique_name.as_bytes());
            unique_name = generate_index_name(&format!("unique_{}", md5));
            unique = format!("CONSTRAINT {} UNIQUE ({})", unique_name, unique_fields);
        }

        // 处理普通索引
        for row in options.table_index.iter() {
            let mut index_fields = String::new();
            let mut index_name = String::new();
            for item in row {
                if index_fields.is_empty() {
                    index_fields = format!("\"{}\"", item);
                    index_name = format!("{}_index_{}", options.table_name, item);
                } else {
                    index_fields = format!("{}, \"{}\"", index_fields, item);
                    index_name = format!("{}_{}", index_name, item);
                }
            }
            let md5 = br_crypto::md5::encrypt_hex(index_name.as_bytes());
            index_name = generate_index_name(&format!("index_{}", md5));
            if index.is_empty() {
                index = format!(
                    "CREATE INDEX {} ON {} ({})",
                    index_name, options.table_name, index_fields
                );
            } else {
                index = format!(
                    "{}\r\nCREATE INDEX {} ON {} ({})",
                    index, index_name, options.table_name, index_fields
                );
            }
        }

        // 生成字段定义
        for (name, field) in options.table_fields.entries() {
            let row = br_fields::field("pgsql", name, field.clone());
            sql = format!("{sql} {row},\r\n");
        }

        // 添加主键
        if !unique.is_empty() {
            sql = sql.trim_end_matches(",\r\n").to_string();
            sql = format!("{sql},\r\n{unique}");
        }
        sql = if sql.trim_end().ends_with(",") {
            format!("{}\r\nPRIMARY KEY(\"{}\")", sql, options.table_key)
        } else {
            format!("{},\r\nPRIMARY KEY(\"{}\")", sql, options.table_key)
        };

        // 生成完整的 CREATE TABLE 语句（修复版：分割执行）
        let create_sql = format!(
            "CREATE TABLE IF NOT EXISTS {} (\r\n{}\r\n)",
            options.table_name, sql
        );

        if self.params.sql {
            // 如果只是查看SQL，返回分割后的语句列表
            let mut statements = vec![create_sql];
            if !index.is_empty() {
                let index_statements: Vec<&str> = index.split("\r\n").filter(|s| !s.trim().is_empty()).collect();
                for index_sql in index_statements {
                    let final_sql = if index_sql.trim().ends_with(';') {
                        index_sql.trim().to_string()
                    } else {
                        format!("{};", index_sql.trim())
                    };
                    statements.push(final_sql);
                }
            }
            let full_sql = statements.join("\r\n");
            println!("DEBUG: Returning SQL in fetch_sql mode: {}", full_sql);
            return JsonValue::from(full_sql);
        }

        // 强制调试输出
        println!("=== TABLE CREATE DEBUG START ===");
        println!("DEBUG: params.sql = {}, table_name = {}", self.params.sql, options.table_name);
        println!("CREATE TABLE SQL: {}", create_sql);
        let (state, data) = self.execute(&create_sql);
        println!("CREATE TABLE result: state={}, data={}", state, data);
        if !state {
            error!("创建表失败: {} - {}", options.table_name, data);
            return JsonValue::from(false);
        }

        // 2. 如果有索引语句，分割执行
        if !index.is_empty() {
            println!("Index statements: {}", index);
            let index_statements: Vec<&str> = index.split("\r\n").filter(|s| !s.trim().is_empty()).collect();
            for (i, index_sql) in index_statements.iter().enumerate() {
                let final_sql = if index_sql.trim().ends_with(';') {
                    index_sql.trim().to_string()
                } else {
                    format!("{};", index_sql.trim())
                };
                println!("Index SQL {}: {}", i+1, final_sql);
                let (idx_state, idx_data) = self.execute(&final_sql);
                println!("Index result {}: state={}, data={}", i+1, idx_state, idx_data);
                if !idx_state {
                    error!("创建索引失败: {} - {}", options.table_name, idx_data);
                    return JsonValue::from(false);
                }
            }
        }

        // 所有操作都成功完成
        JsonValue::from(true)
    }

    fn table_update(&mut self, options: TableOptions) -> JsonValue {
        // 检查表是否存在
        if !self.table_is_exist(&options.table_name) {
            // 表不存在，尝试创建表
            warn!("表 {} 不存在，尝试创建表", options.table_name);
            return self.table_create(options);
        }

        // 清除缓存
        let cache_key = format!("{}{}", self.default, options.table_name);
        if TABLE_FIELDS.lock().unwrap().get(&cache_key).is_some() {
            TABLE_FIELDS.lock().unwrap().remove(&cache_key);
        }

        let mut sql = vec![];
        let fields_list = self.table_info(&options.table_name);
        let mut put = vec![];
        let mut add = vec![];
        let mut del = vec![];

        // 检查需要删除的字段（在新配置中为空）
        for (key, _) in fields_list.entries() {
            if options.table_fields[key].is_empty() {
                del.push(key);
            }
        }

        // 检查需要添加或修改的字段
        for (name, field) in options.table_fields.entries() {
            println!("DEBUG: Processing field '{}', exists in DB: {}", name, !fields_list[name].is_empty());
            let keys: Vec<_> = fields_list.entries().map(|(k, _)| k).collect();
            println!("DEBUG: fields_list keys: {:?}", keys);
            println!("DEBUG: Looking for key '{}' in fields_list: {}", name, keys.contains(&name));
            if !fields_list[name].is_empty() {
                // 字段已存在，检查是否需要更新
                let old_field = &fields_list[name];
                println!("DEBUG: Field '{}' exists, old_field: {:?}", name, old_field);

                // 获取新字段的标题（从 field JSON 对象中）
                let new_title = field["title"].as_str().unwrap_or("");

                // 获取旧字段的注释，并从中提取标题
                // 注释格式通常是 'title|mode|...'，标题是第一部分
                let old_comment = old_field["comment"].as_str().unwrap_or("");
                let old_title = if !old_comment.is_empty() {
                    old_comment.split('|').next().unwrap_or("")
                } else {
                    ""
                };

                // 比较字段名和标题的合并字符串："{字段名}|{标题}"
                let new_field_title = format!("{}|{}", name, new_title);
                let old_field_title = format!("{}|{}", name, old_title);

                // 如果有差异，需要更新
                if new_field_title != old_field_title {
                    put.push(name);
                    continue;
                }

                // 如果字段名和标题都相同，跳过更新
                continue;
            } else {
                // 字段不存在，需要添加
                add.push(name);
            }
        }

        // 如果没有需要添加、删除或修改的字段，直接返回 -1 表示没有变化
        if add.is_empty() && del.is_empty() && put.is_empty() {
            info!(
                "数据库更新情况: {} 成功 更新前检查字段是否有变化，没有变化的不需要更新",
                options.table_name
            );
            return JsonValue::from(-1);
        }

        // 生成 ALTER TABLE 语句 - 改进列存在检查
        for name in add.iter() {
            let name_str = name.to_string();
            
            // 更安全的列存在检查 - 使用更精确的查询
            let column_check_sql = format!(
                "SELECT COUNT(*) as count FROM information_schema.columns WHERE table_schema = 'public' AND table_name = '{}' AND column_name = '{}'",
                options.table_name, name_str
            );
            println!("DEBUG: Checking column existence with SQL: {}", column_check_sql);
            let (check_state, check_result) = self.query(&column_check_sql);
            
            if check_state {
                let count = check_result[0]["count"].as_i64().unwrap_or(0);
                if count > 0 {
                    println!("DEBUG: Column '{}' already exists in table '{}', skipping ADD COLUMN", name_str, options.table_name);
                    continue;
                }
            } else {
                println!("DEBUG: Column check failed for '{}', assuming it doesn't exist", name_str);
            }
            
            let row = br_fields::field("pgsql", &name_str, options.table_fields[name_str.as_str()].clone());
            let alter_sql = format!(
                "ALTER TABLE {} ADD COLUMN {}",
                options.table_name, row
            );
            println!("DEBUG: Adding column with SQL: {}", alter_sql);
            sql.push(alter_sql);
        }

        for name in del.iter() {
            sql.push(format!(
                "ALTER TABLE {} DROP COLUMN \"{}\"",
                options.table_name, name
            ));
        }

        for name in put.iter() {
            let name_str = name.to_string();
            let row = br_fields::field("pgsql", &name_str, options.table_fields[name_str.as_str()].clone());
            println!("DEBUG: br_fields::field returned: {}", row);

            // 解析字段定义，提取类型和注释
            let comment_parts: Vec<&str> = row.split(" comment ").collect();
            let field_type_part = if comment_parts.len() > 1 {
                comment_parts[0].trim()
            } else {
                row.trim()
            };

            // PostgreSQL ALTER COLUMN 需要分别处理类型和注释
            // 提取类型部分（去掉字段名，只保留类型定义）
            let type_def = if let Some(space_pos) = field_type_part.find(' ') {
                &field_type_part[space_pos + 1..]
            } else {
                field_type_part
            };

            // 更新字段类型
            sql.push(format!(
                "ALTER TABLE {} ALTER COLUMN \"{}\" TYPE {}",
                options.table_name, name_str, type_def
            ));

            // 更新注释（如果存在）
            if comment_parts.len() > 1 {
                let comment = comment_parts[1]
                    .trim_start_matches("'")
                    .trim_end_matches("'");
                sql.push(format!(
                    "COMMENT ON COLUMN {}.{} IS '{}'",
                    options.table_name, name_str, comment
                ));
            }

            // 更新默认值和可空性（需要从原始字段定义中解析）
            let old_field = self.table_info(&options.table_name);
            if old_field[name_str.as_str()].is_object() {
                let field_info = &old_field[name_str.as_str()];
                // 检查是否需要更新默认值
                let old_default = field_info["column_default"].as_str().unwrap_or("");
                
                // 如果新的字段定义包含 "not null" 但旧的没有，或者包含不同的默认值
                if row.contains("not null") && !old_default.contains("not null") {
                    sql.push(format!(
                        "ALTER TABLE {} ALTER COLUMN \"{}\" SET NOT NULL",
                        options.table_name, name_str
                    ));
                }
                // 如果包含默认值，处理默认值更新
                if let Some(default_start) = row.find("default '") {
                    if let Some(default_end) = row[default_start + 9..].find('\'') {
                        let new_default = &row[default_start + 9..default_start + 9 + default_end];
                        sql.push(format!(
                            "ALTER TABLE {} ALTER COLUMN \"{}\" SET DEFAULT {}",
                            options.table_name, name_str, new_default
                        ));
                    }
                }
            }
        }

        // 处理唯一索引
        let mut unique_fields = String::new();
        let mut unique_name = String::new();
        for item in options.table_unique.iter() {
            if unique_fields.is_empty() {
                unique_fields = format!("\"{}\"", item);
                unique_name = format!("{}_unique_{}", options.table_name, item);
            } else {
                unique_fields = format!("{}, \"{}\"", unique_fields, item);
                unique_name = format!("{}_{}", unique_name, item);
            }
        }

        if !unique_name.is_empty() {
            let md5 = br_crypto::md5::encrypt_hex(unique_name.as_bytes());
            unique_name = generate_index_name(&format!("unique_{}", md5));

            // 查询现有唯一索引
            let (_, index_list) = self.query(format!(
                "SELECT indexname FROM pg_indexes WHERE tablename = '{}' AND indexdef LIKE '%UNIQUE%'",
                options.table_name
            ).as_str());

            let mut unique_new = vec![];
            for item in index_list.members() {
                if let Some(index_name) = item["indexname"].as_str() {
                    unique_new.push(index_name.to_string());
                }
            }

            // 删除不匹配的唯一索引
            for item in &unique_new {
                if unique_name != *item {
                    sql.push(format!("DROP INDEX IF EXISTS {}", item));
                }
            }

            // 创建新的唯一索引（如果不存在）
            if !unique_new.contains(&unique_name) {
                sql.push(format!(
                    "CREATE UNIQUE INDEX IF NOT EXISTS {} ON {} ({})",
                    unique_name, options.table_name, unique_fields
                ));
            }
        }

        // 处理普通索引
        let mut index_list = vec![];
        for row in options.table_index.iter() {
            let mut index_fields = String::new();
            let mut index_name = String::new();
            for item in row {
                if index_fields.is_empty() {
                    index_fields = format!("\"{}\"", item);
                    index_name = format!("{}_index_{}", options.table_name, item);
                } else {
                    index_fields = format!("{}, \"{}\"", index_fields, item);
                    index_name = format!("{}_{}", index_name, item);
                }
            }
            let md5 = br_crypto::md5::encrypt_hex(index_name.as_bytes());
            index_name = generate_index_name(&format!("index_{}", md5));
            index_list.push(index_name.clone());

            // 查询现有索引
            let (_, existing_indexes) = self.query(
                format!(
                    "SELECT indexname FROM pg_indexes WHERE tablename = '{}'",
                    options.table_name
                )
                .as_str(),
            );

            let mut existing_index_names = vec![];
            for item in existing_indexes.members() {
                if let Some(index_name) = item["indexname"].as_str() {
                    existing_index_names.push(index_name.to_string());
                }
            }

            if !existing_index_names.contains(&index_name) {
                sql.push(format!(
                    "CREATE INDEX IF NOT EXISTS {} ON {} ({})",
                    index_name, options.table_name, index_fields
                ));
            }
        }

        // 删除不再需要的索引
        let (_, all_indexes) = self.query(
            format!(
                "SELECT indexname FROM pg_indexes WHERE tablename = '{}'",
                options.table_name
            )
            .as_str(),
        );

        for item in all_indexes.members() {
            if let Some(index_name) = item["indexname"].as_str() {
                // 跳过主键索引和唯一索引（它们由上面的逻辑处理）
                if !index_name.contains("_pkey") && !index_list.contains(&index_name.to_string()) {
                    // 检查是否是表相关的索引
                    if index_name.starts_with(&format!("{}_", options.table_name)) {
                        sql.push(format!("DROP INDEX IF EXISTS {}", index_name));
                    }
                }
            }
        }

        if self.params.sql {
            return JsonValue::from(sql.join("\r\n"));
        }

        // 如果没有需要执行的 SQL，返回 -1 表示没有变化
        if sql.is_empty() {
            info!(
                "数据库更新情况: {} 成功 更新前检查字段是否有变化，没有变化的不需要更新",
                options.table_name
            );
            return JsonValue::from(-1);
        }

        // 执行 SQL 语句
        for item in sql.iter() {
            let (state, res) = self.execute(item.as_str());
            match state {
                true => {}
                false => {
                    error!("{} 更新失败: {} \r\n {}", options.table_name, item, res);
                    return JsonValue::from(0);
                }
            }
        }

        JsonValue::from(1)
    }

    fn table_info(&mut self, table: &str) -> JsonValue {
        let cache_key = format!("{}{}", self.default, table);
        {
            let fields = TABLE_FIELDS.lock().unwrap();
            if let Some(cached) = fields.get(&cache_key) {
                return cached.clone();
            }
        }

        // 查询 PostgreSQL 表结构信息
        let sql = format!(
            "SELECT 
                COL.COLUMN_NAME,
                COL.DATA_TYPE,
                COL.UDT_NAME,
                COL.CHARACTER_MAXIMUM_LENGTH,
                COL.NUMERIC_PRECISION,
                COL.NUMERIC_SCALE,
                COL.COLUMN_DEFAULT,
                COL.IS_NULLABLE,
                COALESCE(DESCRIPTION.DESCRIPTION, '') AS COMMENT
            FROM INFORMATION_SCHEMA.COLUMNS COL
            LEFT JOIN pg_catalog.pg_description DESCRIPTION
                ON DESCRIPTION.objsubid = COL.ORDINAL_POSITION
                AND DESCRIPTION.objoid = (
                    SELECT oid FROM pg_catalog.pg_class 
                    WHERE relname = COL.TABLE_NAME 
                    AND relnamespace = (SELECT oid FROM pg_catalog.pg_namespace WHERE nspname = 'public')
                    LIMIT 1
                )
            WHERE COL.TABLE_SCHEMA = 'public' 
                AND COL.TABLE_NAME = '{}'",
            table
        );

        println!("DEBUG: Table info SQL: {}", sql);
        let (state, data) = self.query(sql.as_str());
        println!("DEBUG: Table info query result: state={}, data={}", state, data);
        let mut list = object! {};

        if state {
            println!("DEBUG: Processing {} data items", data.members().len());
            for (i, item) in data.members().enumerate() {
                println!("DEBUG: Item {}: {}", i, item);
                if let Some(field_name) = item["column_name"].as_str() {
                    println!("DEBUG: Extracted field_name: '{}'", field_name);
                    let mut row = object! {};
                    row["field"] = JsonValue::from(field_name);
                    row["type"] = item["data_type"].clone();
                    row["udt_name"] = item["udt_name"].clone();
                    row["comment"] = item["comment"].clone();
                    row["is_nullable"] = item["is_nullable"].clone();
                    row["column_default"] = item["column_default"].clone();
                    row["character_maximum_length"] = item["character_maximum_length"].clone();
                    row["numeric_precision"] = item["numeric_precision"].clone();
                    row["numeric_scale"] = item["numeric_scale"].clone();
                    list[field_name] = row;
                    println!("DEBUG: Stored field '{}' in list", field_name);
                } else {
                    println!("DEBUG: Could not extract COLUMN_NAME from item: {}", item);
                }
            }
            let list_clone = list.clone();
            println!("DEBUG: Storing {} fields in cache for key: {}", list_clone.len(), cache_key);
            TABLE_FIELDS.lock().unwrap().insert(cache_key, list_clone);
        }
        list
    }

    fn table_is_exist(&mut self, name: &str) -> bool {
        let sql = format!(
            "SELECT COUNT(*) as count FROM information_schema.tables WHERE table_schema = 'public' AND table_name = '{}'",
            name
        );
        let (state, data) = self.query(sql.as_str());
        match state {
            true => {
                if !data.is_empty() {
                    if let Some(count) = data[0]["count"].as_i64() {
                        return count > 0;
                    }
                }
                false
            }
            false => false,
        }
    }

    fn table(&mut self, name: &str) -> &mut Pgsql {
        self.params.table = name.to_string();
        self
    }

    fn change_table(&mut self, name: &str) -> &mut Self {
        self.params.table = name.to_string();
        self
    }

    fn autoinc(&mut self) -> &mut Self {
        self
    }

    fn fetch_sql(&mut self) -> &mut Self {
        self.params.sql = true;
        self
    }

    fn order(&mut self, _field: &str, _by: bool) -> &mut Self {
        self
    }

    fn group(&mut self, _field: &str) -> &mut Self {
        self
    }

    fn distinct(&mut self) -> &mut Self {
        self
    }

    fn json(&mut self, _field: &str) -> &mut Self {
        self
    }

    fn location(&mut self, _field: &str) -> &mut Self {
        self
    }

    fn field(&mut self, _field: &str) -> &mut Self {
        self
    }

    fn hidden(&mut self, _name: &str) -> &mut Self {
        self
    }

    fn where_and(&mut self, _field: &str, _compare: &str, _value: JsonValue) -> &mut Self {
        self
    }

    fn where_or(&mut self, _field: &str, _compare: &str, _value: JsonValue) -> &mut Self {
        self
    }

    fn where_column(&mut self, _field_a: &str, _compare: &str, _field_b: &str) -> &mut Self {
        self
    }

    fn update_column(&mut self, _field_a: &str, _compare: &str) -> &mut Self {
        self
    }

    fn page(&mut self, _page: i32, _limit: i32) -> &mut Self {
        self
    }

    fn column(&mut self, _field: &str) -> JsonValue {
        JsonValue::Null
    }

    fn count(&mut self) -> JsonValue {
        JsonValue::from(0)
    }

    fn max(&mut self, _field: &str) -> JsonValue {
        JsonValue::from(0)
    }

    fn min(&mut self, _field: &str) -> JsonValue {
        JsonValue::from(0)
    }

    fn sum(&mut self, _field: &str) -> JsonValue {
        JsonValue::from(0)
    }

    fn avg(&mut self, _field: &str) -> JsonValue {
        JsonValue::from(0)
    }

    fn select(&mut self) -> JsonValue {
        array![]
    }

    fn find(&mut self) -> JsonValue {
        object! {}
    }

    fn value(&mut self, _field: &str) -> JsonValue {
        JsonValue::Null
    }

    fn insert(&mut self, mut data: JsonValue) -> JsonValue {
        let thread_id = format!("{:?}", thread::current().id());

        let fields_list = self.table_info(&self.params.table.clone());

        let mut fields = vec![];
        let mut values = vec![];

        // 处理自动递增 ID
        if !self.params.autoinc && data["id"].is_empty() {
            data["id"] =
                format!("{:X}", chrono::Local::now().timestamp_nanos_opt().unwrap()).into();
        }

        for (field, value) in data.entries() {
            fields.push(format!("\"{}\"", field));

            if self.params.location.has_key(field) && !self.params.location[field].is_empty() {
                if value.is_empty() {
                    values.push("NULL".to_string());
                    continue;
                }
                let comment = fields_list[field]["comment"].to_string();
                let srid = comment
                    .split("|")
                    .collect::<Vec<&str>>()
                    .last()
                    .unwrap_or(&"4326")
                    .to_string();
                let location = value.to_string().replace(",", " ");
                values.push(format!("ST_GeomFromText('POINT({})',{})", location, srid));
                continue;
            }

            if value.is_string() || value.is_array() || value.is_object() {
                values.push(format!("'{}'", value.to_string().replace("'", "''")));
                continue;
            } else if value.is_number() || value.is_boolean() || value.is_null() {
                values.push(format!("{}", value));
                continue;
            } else {
                values.push(format!("'{}'", value));
                continue;
            }
        }

        let fields = fields.join(",");
        let values = values.join(",");

        let sql = format!(
            "INSERT INTO {} ({}) VALUES ({});",
            self.params.table, fields, values
        );

        if self.params.sql {
            return JsonValue::from(sql);
        }

        let (state, ids) = self.execute(sql.as_str());
        match state {
            true => match self.params.autoinc {
                true => ids.clone(),
                false => data["id"].clone(),
            },
            false => {
                let error_msg = ids
                    .as_str()
                    .map(|s| s.to_string())
                    .unwrap_or_else(|| ids.to_string());
                error!(
                    "插入记录失败 - 线程ID: {} | 数据库配置: {} | 数据库名: {} | 表名: {} | SQL语句: [{}] | 错误详情: {}",
                    thread_id,
                    self.default,
                    self.connection.database,
                    self.params.table,
                    sql,
                    error_msg
                );
                JsonValue::from("")
            }
        }
    }

    fn insert_all(&mut self, mut data: JsonValue) -> JsonValue {
        let thread_id = format!("{:?}", thread::current().id());
        let list = array![];

        if !data.is_array() {
            return list;
        }

        let fields_list = self.table_info(&self.params.table.clone());
        let mut fields = vec![];

        // 获取所有记录的字段列表的并集
        use std::collections::HashSet;
        let mut field_set = HashSet::new();
        for record in data.members() {
            for (field_name, _) in record.entries() {
                field_set.insert(field_name.to_string());
            }
        }

        // 转换为排序的字段列表（确保顺序一致性）
        let mut field_vec: Vec<String> = field_set.into_iter().collect();
        field_vec.sort();
        for field_name in &field_vec {
            fields.push(format!("\"{}\"", field_name));
        }

        // 处理自动递增 ID
        if !self.params.autoinc {
            for i in 0..data.len() {
                if data[i]["id"].is_empty() {
                    data[i]["id"] = format!(
                        "{:X}",
                        chrono::Local::now().timestamp_nanos_opt().unwrap() + i as i64
                    )
                    .into();
                }
            }
        }

        let mut values_list = vec![];
        for record in data.members() {
            let mut values = vec![];

            // 按照字段列表的顺序构建值
            for field_name in &field_vec {
                let value = record[field_name].clone();

                if self.params.location.has_key(field_name)
                    && !self.params.location[field_name].is_empty()
                {
                    if value.is_empty() {
                        values.push("NULL".to_string());
                        continue;
                    }
                    let comment = fields_list[field_name]["comment"].to_string();
                    let srid = comment
                        .split("|")
                        .collect::<Vec<&str>>()
                        .last()
                        .unwrap_or(&"4326")
                        .to_string();
                    let location = value.to_string().replace(",", " ");
                    values.push(format!("ST_GeomFromText('POINT({})',{})", location, srid));
                    continue;
                }

                if value.is_string() || value.is_array() || value.is_object() {
                    values.push(format!("'{}'", value.to_string().replace("'", "''")));
                } else if value.is_number() || value.is_boolean() || value.is_null() {
                    values.push(format!("{}", value));
                } else {
                    values.push(format!("'{}'", value.to_string().replace("'", "''")));
                }
            }

            values_list.push(format!("({})", values.join(",")));
        }

        let fields = fields.join(",");
        let values = values_list.join(",");

        let sql = format!(
            "INSERT INTO {} ({}) VALUES {};",
            self.params.table, fields, values
        );

        if self.params.sql {
            return JsonValue::from(sql);
        }

        // 调试信息：确保表名正确
        if self.params.table.is_empty() {
            error!("批量插入失败：表名为空 - 线程ID: {}", thread_id);
            return JsonValue::from("");
        }

        let (state, result) = self.execute(sql.as_str());
        match state {
            true => {
                // 对于批量插入，如果没有自动递增ID，返回插入的记录数
                if self.params.autoinc {
                    result.clone()
                } else {
                    // 返回插入的记录数量
                    JsonValue::from(data.len() as i64)
                }
            }
            false => {
                let error_msg = result
                    .as_str()
                    .map(|s| s.to_string())
                    .unwrap_or_else(|| result.to_string());
                error!(
                    "批量插入失败 - 线程ID: {} | 数据库配置: {} | 数据库名: {} | 表名: {} | SQL语句: [{}] | 错误详情: {}",
                    thread_id,
                    self.default,
                    self.connection.database,
                    self.params.table,
                    sql,
                    error_msg
                );
                JsonValue::from("")
            }
        }
    }

    fn update(&mut self, _data: JsonValue) -> JsonValue {
        JsonValue::from(0)
    }

    fn update_all(&mut self, _data: JsonValue) -> JsonValue {
        JsonValue::from(0)
    }

    fn delete(&mut self) -> JsonValue {
        JsonValue::from(0)
    }

    fn inc(&mut self, _field: &str, _num: f64) -> &mut Self {
        self
    }

    fn dec(&mut self, _field: &str, _num: f64) -> &mut Self {
        self
    }

    fn buildsql(&mut self) -> String {
        String::new()
    }

    fn join_fields(&mut self, _fields: Vec<&str>) -> &mut Self {
        self
    }

    fn join(
        &mut self,
        _main_table: &str,
        _main_fields: &str,
        _right_table: &str,
        _right_fields: &str,
    ) -> &mut Self {
        self
    }

    fn join_inner(&mut self, _table: &str, _main_fields: &str, _second_fields: &str) -> &mut Self {
        self
    }
}