remdb 0.3.2

#![cfg_attr(not(feature = "std"), no_std)]
#![recursion_limit = "1024"]
#![allow(
    static_mut_refs,
    clippy::not_unsafe_ptr_arg_deref,
    clippy::never_loop,
    clippy::absurd_extreme_comparisons,
    clippy::mut_from_ref
)]

use crate::table::Defer;
use core::ptr::NonNull;

// 导出公共API
pub mod compression;
pub mod config;
#[cfg(feature = "ha")]
pub mod ha;
pub mod index;
pub mod json;
#[cfg(feature = "log")]
pub mod log;
pub mod memory;
pub mod model;
pub mod monitor;
pub mod platform;
#[cfg(feature = "pubsub")]
pub mod pubsub;
pub mod rbac;
pub mod sql;
pub mod sync;
pub mod system_tables;
pub mod table;
pub mod time_series;
pub mod transaction;
pub mod types;
pub mod utf8;
pub mod wal_compression;

/// 安全地获取 Mutex 锁，自动处理中毒情况
/// 不会 panic
#[macro_export]
macro_rules! try_lock {
    ($lock:expr) => {{
        #[cfg(feature = "std")]
        {
            $lock.lock().unwrap_or_else(|e| e.into_inner())
        }
        #[cfg(not(feature = "std"))]
        {
            $lock.lock().unwrap_or_else(|_| unreachable!())
        }
    }};
}

/// 安全地获取 RwLock 读锁，自动处理中毒情况
/// 不会 panic
#[macro_export]
macro_rules! try_read {
    ($lock:expr) => {{
        #[cfg(feature = "std")]
        {
            $lock.read().unwrap_or_else(|e| e.into_inner())
        }
        #[cfg(not(feature = "std"))]
        {
            // no_std 下没有真正的 RwLock，使用 Mutex 的 lock() 模拟
            $lock.lock().unwrap_or_else(|_| unreachable!())
        }
    }};
}

/// 安全地获取 RwLock 写锁，自动处理中毒情况
/// 不会 panic
#[macro_export]
macro_rules! try_write {
    ($lock:expr) => {{
        #[cfg(feature = "std")]
        {
            $lock.write().unwrap_or_else(|e| e.into_inner())
        }
        #[cfg(not(feature = "std"))]
        {
            // no_std 下没有真正的 RwLock，使用 Mutex 的 lock() 模拟
            $lock.lock().unwrap_or_else(|_| unreachable!())
        }
    }};
}

// 导出核心类型
pub use compression::CompressionScheme;
pub use rbac::{Permission, RbacManager, Role, User};
pub use system_tables::{
    get_query_resource_config, get_vector_compression_config, init_system_tables,
};
pub use table::{MemoryTable, RecordCursor, RecordIdCursor, RecordRef};
pub use types::{
    DataType, DistanceType, FieldDef, IndexType, RecordStatus, RemDbError, Result, TableDef, Value,
    VectorIndexType, VectorMetadata, DEFAULT_JSON_SIZE, DEFAULT_TEXT_SIZE, MAX_STRING_LEN,
    MAX_TEXT_LEN,
};

pub use index::{
    init_index_build_thread_pool, AnySecondaryIndex, BTreeIndex, IndexStats, PrimaryIndex,
    PrimaryIndexItem, SecondaryIndex, TTreeIndex,
};
pub use monitor::{DbMetrics, DbMetricsSnapshot, HealthCheckResult, HealthStatus};
pub use time_series::{
    CompressionType, TimeSeriesConfig, TimeSeriesIndex, TimeSeriesRecord, TimeSeriesTable,
    TimeSeriesTableDef,
};
pub use transaction::{Transaction, TransactionType};

// 重新导出宏
pub use remdb_macros::{database, table, MemdbTable};

// 引入alloc模块
extern crate alloc;
use alloc::boxed::Box;
use alloc::string::String;
use alloc::string::ToString;
use alloc::sync::Arc;
use alloc::vec::Vec;

#[cfg(feature = "log")]
use crate::log::{debug, error, info, warn};

#[cfg(feature = "log")]
pub use crate::log::{init_logger, init_logger_with_file};

/// 字段约束信息
#[derive(Clone, Debug)]
pub struct FieldConstraint {
    /// 是否为主键
    pub primary_key: bool,
    /// 是否非空
    pub not_null: bool,
    /// 是否唯一
    pub unique: bool,
    /// 是否自增
    pub auto_increment: bool,
}

/// ALTER TABLE操作类型
#[derive(Clone, Debug)]
pub enum AlterTableOperation {
    /// 添加新列
    AddColumn {
        name: alloc::string::String,
        data_type: DataType,
        size: u16,
        distance_type: Option<DistanceType>,
        default_value: Option<Value>,
        constraints: FieldConstraint,
    },
    /// 删除列
    DropColumn { name: alloc::string::String },
    /// 修改列
    ModifyColumn {
        name: alloc::string::String,
        data_type: DataType,
        size: u16,
        distance_type: Option<DistanceType>,
        default_value: Option<Value>,
        constraints: FieldConstraint,
    },
    /// 重命名列
    RenameColumn {
        old_name: alloc::string::String,
        new_name: alloc::string::String,
    },
}

/// DDL执行器trait，定义创建表、索引、时序表和修改表的方法
pub trait DdlExecutor {
    /// 创建表
    fn create_table(
        &mut self,
        name: &str,
        fields: &[(&str, DataType, u16, Option<DistanceType>, Option<Value>)],
        constraints: Option<&[FieldConstraint]>,
        primary_key: Option<Vec<usize>>,
        max_records: Option<usize>,
    ) -> Result<()>;

    /// 修改表结构
    fn alter_table(&mut self, table_name: &str, operation: AlterTableOperation) -> Result<()>;

    /// 创建索引
    fn create_index(
        &mut self,
        table_name: &str,
        field_name: &str,
        index_type: IndexType,
    ) -> Result<()>;

    /// 创建时序表
    fn create_time_series_table(
        &mut self,
        name: &str,
        time_field: &str,
        value_field: &str,
        tag_fields: &[&str],
        config: Option<TimeSeriesConfig>,
    ) -> Result<()>;
}

/// 数据库实例
pub struct RemDb {
    /// 数据库名称
    pub name: String,
    /// 数据库配置
    pub config: &'static config::DbConfig,
    /// 内存表数组
    tables: Vec<Option<MemoryTable>>,
    /// 时序表数组
    time_series_tables: Vec<Option<time_series::TimeSeriesTable>>,
    /// 主键索引数组
    primary_indices: Vec<Option<PrimaryIndex>>,
    /// 辅助索引数组
    secondary_indices: Vec<Option<AnySecondaryIndex>>,
    /// 是否处于低功耗模式
    low_power_mode: bool,
    /// 低功耗模式下的内存使用限制
    low_power_memory_limit: usize,
    /// 全局快照版本号
    pub snapshot_version: u32,
    /// 数据库监控指标
    pub metrics: monitor::DbMetrics,
    /// 数据库状态
    pub status: DatabaseStatus,
    /// 模型管理器
    pub model_manager: model::ModelManager,
    /// RBAC管理器
    pub rbac_manager: rbac::RbacManager,
    /// 数据库管理器，用于管理多个数据库实例
    database_manager: DatabaseManager,
}

/// 数据库状态
#[derive(Clone, Debug, PartialEq)]
pub enum DatabaseStatus {
    /// 已创建
    Created,
    /// 已打开
    Open,
    /// 已关闭
    Closed,
    /// 已删除
    Dropped,
}

/// 数据库信息结构体
#[derive(Clone, Debug, PartialEq)]
pub struct DatabaseInfo {
    /// 数据库名称
    pub name: String,
    /// 数据库类型
    pub database_type: String,
    /// 数据库状态
    pub status: DatabaseStatus,
    /// 表数量
    pub table_count: usize,
    /// 内存使用情况（字节）
    pub memory_usage: usize,
}

/// 数据库配置
#[derive(Clone, Debug)]
pub struct DatabaseConfig {
    /// 数据库名称
    pub name: String,
    /// 内存限制
    pub memory_limit: Option<usize>,
    /// 最大表数量
    pub max_tables: Option<usize>,
    /// WAL模式
    pub wal_mode: Option<String>,
    /// 默认索引类型
    pub default_index_type: Option<IndexType>,
    /// 临时存储位置
    pub temp_store: Option<String>,
}

/// 数据库管理器
pub struct DatabaseManager {
    /// 数据库实例列表
    databases: Vec<Arc<RemDb>>,
    /// 当前活跃数据库
    current_database: Option<usize>,
    /// 最大数据库实例数
    max_databases: usize,
}

impl DatabaseManager {
    /// 创建新的数据库管理器
    pub fn new(max_databases: usize) -> Self {
        Self {
            databases: Vec::with_capacity(max_databases),
            current_database: None,
            max_databases,
        }
    }

    /// 获取当前数据库
    pub fn get_current_database(&self) -> Option<Arc<RemDb>> {
        self.current_database.map(|idx| self.databases[idx].clone())
    }

    /// 切换到指定数据库
    pub fn use_database(&mut self, name: &str) -> Result<()> {
        if let Some(idx) = self.databases.iter().position(|db| db.name == name) {
            // 新建的数据库状态为 Created（尚未关闭），同样可以使用
            if self.databases[idx].status == DatabaseStatus::Open
                || self.databases[idx].status == DatabaseStatus::Created
            {
                self.current_database = Some(idx);
                Ok(())
            } else {
                Err(RemDbError::DatabaseClosed)
            }
        } else {
            Err(RemDbError::DatabaseNotFound)
        }
    }

    /// 创建新数据库
    pub fn create_database(
        &mut self,
        name: &str,
        _schema: &str,
        config: Option<DatabaseConfig>,
    ) -> Result<Arc<RemDb>> {
        // 检查数据库是否已存在
        if self.databases.iter().any(|db| db.name == name) {
            return Err(RemDbError::DatabaseExists);
        }

        // 检查是否达到最大数据库实例数
        if self.databases.len() >= self.max_databases {
            return Err(RemDbError::MaxDatabasesReached);
        }

        // 获取默认内存分配器
        use crate::config::DefaultMemoryAllocator;

        // 创建数据库配置
        let db_config = Box::leak(Box::new(config::DbConfig {
            tables: vec![],
            total_memory: config
                .as_ref()
                .and_then(|c| c.memory_limit)
                .unwrap_or(1024 * 1024 * 1024), // 默认1GB
            default_max_records: 100000,
            low_power_mode_supported: true,
            low_power_max_records: Some(10000),
            memory_allocator: &DefaultMemoryAllocator,
            wal_config: config::WALConfig {
                log_path: Box::leak(format!("./data/{}", name).into_boxed_str()),
                log_mode: crate::config::LogMode::Async,
                checkpoint_interval_ms: 60000,
                log_file_size_limit: 16 * 1024 * 1024,
                log_prealloc_size: 0,
                log_segment_size: 16 * 1024 * 1024,
                retained_checkpoints: 2,
                max_consecutive_invalid: 100,
                skip_threshold: 20,
                skip_block_size: 4096,
                max_skip_attempts: 10,
                compression_type: crate::config::WALCompressionType::None,
                compression_level: 3,
            },
            time_series_defaults: crate::time_series::TimeSeriesConfig {
                max_partitions: 100,
                partition_duration_secs: 3600,
                retention_period_secs: 86400 * 30,
                compression: crate::time_series::CompressionType::None,
            },
            #[cfg(feature = "pubsub")]
            pubsub_config: None,
            #[cfg(feature = "ha")]
            ha_config: None,
            model_worker_config: config::ModelWorkerConfig::default(),
        }));

        // 创建数据库实例
        let db = Arc::new(RemDb::new_with_name(name, db_config));

        // 记录CreateDatabase操作的WAL日志
        unsafe {
            if let Some(log_manager) = crate::transaction::get_log_manager() {
                let mut new_data = alloc::vec::Vec::new();
                new_data.push(name.len() as u8);
                new_data.extend_from_slice(name.as_bytes());

                let new_data_size = new_data.len() as u16;

                let mut var_log_item = crate::transaction::VariableSizeLogItem {
                    header: crate::transaction::LogItem {
                        op_type: crate::transaction::LogOperation::CreateDatabase,
                        table_id: 0,
                        record_id: 0,
                        old_data_size: 0,
                        new_data_size,
                        tx_id: 0,
                        timestamp: crate::platform::get_timestamp_us(),
                        checksum: 0,
                    },
                    old_data: Vec::new(),
                    new_data,
                };

                let calculated_checksum =
                    crate::transaction::Transaction::calculate_variable_size_log_item_checksum(
                        &var_log_item,
                    );
                var_log_item.header.checksum = calculated_checksum;

                let _ = log_manager.write_variable_size_log_item(&var_log_item);
            }
        }

        // 添加到数据库列表
        self.databases.push(db.clone());

        Ok(db)
    }

    /// 关闭数据库
    pub fn close_database(&mut self, name: &str) -> Result<()> {
        if let Some(_idx) = self.databases.iter().position(|db| db.name == name) {
            // 这里需要实现关闭数据库的逻辑
            // 例如：持久化数据、释放资源等
            Ok(())
        } else {
            Err(RemDbError::DatabaseNotFound)
        }
    }

    /// 删除数据库
    pub fn drop_database(&mut self, name: &str) -> Result<()> {
        if let Some(idx) = self.databases.iter().position(|db| db.name == name) {
            // 这里需要实现删除数据库的逻辑
            // 例如：删除持久化文件、释放所有资源等
            self.databases.remove(idx);
            if let Some(current) = self.current_database {
                if current >= idx {
                    self.current_database = Some(current - 1);
                }
            }
            Ok(())
        } else {
            Err(RemDbError::DatabaseNotFound)
        }
    }

    /// 获取所有数据库信息
    pub fn list_databases(&self) -> Result<Vec<DatabaseInfo>> {
        let mut databases_info = Vec::new();

        for db in &self.databases {
            // 计算表数量（排除None值）
            let table_count = db.tables.iter().filter(|table| table.is_some()).count();

            // 计算内存使用情况
            let memory_usage = db
                .metrics
                .used_memory
                .load(core::sync::atomic::Ordering::Relaxed);

            // 创建数据库信息
            let db_info = DatabaseInfo {
                name: db.name.clone(),
                database_type: "RemDb".to_string(),
                status: db.status.clone(),
                table_count,
                memory_usage,
            };

            databases_info.push(db_info);
        }

        Ok(databases_info)
    }
}

// 为RemDb实现Send和Sync trait
// 注意：这是安全的，因为RemDb的所有字段都是线程安全的
unsafe impl Send for RemDb {}
unsafe impl Sync for RemDb {}

// 实现Drop trait，确保资源正确释放
impl Drop for RemDb {
    fn drop(&mut self) {
        // 关闭HA管理器
        #[cfg(feature = "ha")]
        if let Err(_e) = crate::ha::shutdown() {
            // 关闭失败，记录错误但不影响程序退出
        }
    }
}

impl RemDb {
    /// 快照魔数
    const SNAPSHOT_MAGIC: u32 = 0x52454D44; // 'REMD'
    /// 快照版本
    const SNAPSHOT_VERSION: u32 = 1;

    /// 创建新的数据库实例
    pub fn new(config: &'static config::DbConfig) -> Self {
        Self::new_with_name("default", config)
    }

    /// 创建带有名称的数据库实例
    pub fn new_with_name(name: &str, config: &'static config::DbConfig) -> Self {
        // 计算低功耗模式下的内存限制（如果启用）
        let low_power_memory_limit = if config.low_power_mode_supported {
            // 低功耗模式下，内存使用限制为正常模式的50%
            (config.total_memory / 2).max(1024 * 1024) // 至少1MB
        } else {
            config.total_memory
        };

        // 初始化监控指标
        let metrics = monitor::DbMetrics::new(config.total_memory);

        // 初始化表和索引数组，并预分配足够的容量，避免后续内存重新分配
        let tables = Vec::with_capacity(config.tables.len());
        let time_series_tables = Vec::new();
        let primary_indices = Vec::with_capacity(config.tables.len());
        let secondary_indices = Vec::with_capacity(config.tables.len());

        // 初始化数据库管理器，默认最大10个数据库实例
        let database_manager = DatabaseManager::new(10);

        RemDb {
            name: name.to_string(),
            config,
            tables,
            time_series_tables,
            primary_indices,
            secondary_indices,
            low_power_mode: false, // 默认不启用低功耗模式
            low_power_memory_limit,
            snapshot_version: 0, // 初始快照版本为0
            metrics,
            status: DatabaseStatus::Created,
            model_manager: model::ModelManager::new(),
            rbac_manager: rbac::RbacManager::new(),
            database_manager,
        }
    }

    /// 创建新数据库
    pub fn create_database(&mut self, name: &str) -> Result<()> {
        // 检查数据库名称是否有效
        if name.is_empty() {
            return Err(RemDbError::ConfigError);
        }

        // 使用 DatabaseManager 创建新数据库
        // 传递空字符串作为 schema，使用默认配置
        let config = Some(DatabaseConfig {
            name: name.to_string(),
            memory_limit: Some(self.config.total_memory),
            max_tables: Some(100), // 默认最大100个表
            wal_mode: Some("async".to_string()),
            default_index_type: Some(IndexType::SortedArray),
            temp_store: Some("./tmp".to_string()),
        });

        // 调用 DatabaseManager 的 create_database 方法
        match self.database_manager.create_database(name, "", config) {
            Ok(_db) => {
                // 数据库创建成功
                Ok(())
            }
            Err(e) => {
                // 数据库创建失败
                Err(e)
            }
        }
    }

    /// 使用指定数据库
    pub fn use_database(&mut self, name: &str) -> Result<()> {
        // 检查数据库名称是否有效
        if name.is_empty() {
            return Err(RemDbError::ConfigError);
        }

        // 使用 DatabaseManager 切换到指定数据库
        match self.database_manager.use_database(name) {
            Ok(_) => {
                // 数据库切换成功
                Ok(())
            }
            Err(e) => {
                // 数据库切换失败
                Err(e)
            }
        }
    }

    /// 关闭指定数据库
    pub fn close_database(&mut self, name: &str) -> Result<()> {
        // 检查数据库名称是否有效
        if name.is_empty() {
            return Err(RemDbError::ConfigError);
        }

        // 使用 DatabaseManager 关闭指定数据库
        match self.database_manager.close_database(name) {
            Ok(_) => {
                // 数据库关闭成功
                Ok(())
            }
            Err(e) => {
                // 数据库关闭失败
                Err(e)
            }
        }
    }

    /// 删除指定数据库
    pub fn drop_database(&mut self, name: &str) -> Result<()> {
        // 检查数据库名称是否有效
        if name.is_empty() {
            return Err(RemDbError::ConfigError);
        }

        // 使用 DatabaseManager 删除指定数据库
        match self.database_manager.drop_database(name) {
            Ok(_) => {
                // 数据库删除成功
                Ok(())
            }
            Err(e) => {
                // 数据库删除失败
                Err(e)
            }
        }
    }

    /// 创建角色
    pub fn create_role(&mut self, role_name: &str) -> Result<()> {
        self.rbac_manager
            .create_role(role_name.to_string())
            .map_err(|_e| RemDbError::ConfigError)?;

        // 保存到系统表
        unsafe {
            let _ = RbacManager::save_to_system_tables(&mut *self);
        }

        Ok(())
    }

    /// 删除角色
    pub fn drop_role(&mut self, role_name: &str) -> Result<()> {
        self.rbac_manager
            .drop_role(role_name)
            .map_err(|_e| RemDbError::ConfigError)?;

        // 保存到系统表
        unsafe {
            let _ = RbacManager::save_to_system_tables(&mut *self);
        }

        Ok(())
    }

    /// 授予权限给角色
    pub fn grant_permission(
        &mut self,
        role_name: &str,
        permission: rbac::Permission,
        table_name: Option<String>,
        column_name: Option<String>,
    ) -> Result<()> {
        self.rbac_manager
            .grant_permission(role_name, permission, table_name, column_name)
            .map_err(|_e| RemDbError::ConfigError)?;

        // 保存到系统表
        unsafe {
            let _ = RbacManager::save_to_system_tables(&mut *self);
        }

        Ok(())
    }

    /// 撤销角色的权限
    pub fn revoke_permission(
        &mut self,
        role_name: &str,
        permission: &rbac::Permission,
        table_name: &Option<String>,
        column_name: &Option<String>,
    ) -> Result<()> {
        self.rbac_manager
            .revoke_permission(role_name, permission, table_name, column_name)
            .map_err(|_e| RemDbError::ConfigError)?;

        // 保存到系统表
        unsafe {
            let _ = RbacManager::save_to_system_tables(&mut *self);
        }

        Ok(())
    }

    /// 创建用户
    pub fn create_user(&mut self, user_name: &str) -> Result<()> {
        self.rbac_manager
            .create_user(user_name.to_string())
            .map_err(|_e| RemDbError::ConfigError)?;

        // 保存到系统表
        unsafe {
            let _ = RbacManager::save_to_system_tables(&mut *self);
        }

        Ok(())
    }

    /// 删除用户
    pub fn drop_user(&mut self, user_name: &str) -> Result<()> {
        self.rbac_manager
            .drop_user(user_name)
            .map_err(|_e| RemDbError::ConfigError)?;

        // 保存到系统表
        unsafe {
            let _ = RbacManager::save_to_system_tables(&mut *self);
        }

        Ok(())
    }

    /// 授予角色给用户
    pub fn grant_role(&mut self, user_name: &str, role_name: &str) -> Result<()> {
        self.rbac_manager
            .grant_role(user_name, role_name)
            .map_err(|_e| RemDbError::ConfigError)?;

        // 保存到系统表
        unsafe {
            let _ = RbacManager::save_to_system_tables(&mut *self);
        }

        Ok(())
    }

    /// 撤销用户的角色
    pub fn revoke_role(&mut self, user_name: &str, role_name: &str) -> Result<()> {
        self.rbac_manager
            .revoke_role(user_name, role_name)
            .map_err(|_e| RemDbError::ConfigError)?;

        // 保存到系统表
        unsafe {
            let _ = RbacManager::save_to_system_tables(&mut *self);
        }

        Ok(())
    }

    /// 检查用户是否有特定权限
    pub fn check_permission(
        &self,
        user_name: &str,
        permission: &rbac::Permission,
        table_name: &Option<String>,
        column_name: &Option<String>,
    ) -> Result<bool> {
        self.rbac_manager
            .check_permission(user_name, permission, table_name, column_name)
            .map_err(|_e| RemDbError::ConfigError)
    }

    /// 获取当前系统中可用的数据库列表
    pub fn databases(&self) -> Result<Vec<DatabaseInfo>> {
        // 使用 DatabaseManager 获取所有数据库的信息
        // 由于 DatabaseManager::list_databases 只需要不可变引用，我们可以直接调用
        self.database_manager.list_databases()
    }

    /// 获取表
    pub fn get_table(&self, table_id: usize) -> Result<&MemoryTable> {
        if table_id >= self.tables.len() {
            return Err(RemDbError::RecordNotFound);
        }

        match &self.tables[table_id] {
            Some(table) => Ok(table),
            None => Err(RemDbError::RecordNotFound),
        }
    }

    /// 根据ID获取记录引用（零拷贝）
    pub fn get_by_id_ref(&self, table_id: usize, id: usize) -> Result<Option<RecordRef<'_>>> {
        let table = self.get_table(table_id)?;
        Ok(table.get_by_id_ref(id))
    }

    /// 扫描游标（零拷贝）
    pub fn scan_ref(&self, table_id: usize) -> Result<RecordCursor<'_>> {
        let table = self.get_table(table_id)?;
        Ok(table.scan_ref())
    }

    /// 辅助索引范围查询（零拷贝）
    pub fn get_by_index_ref(
        &mut self,
        table_id: usize,
        start_key: *const u8,
        start_key_size: usize,
        end_key: *const u8,
        end_key_size: usize,
    ) -> Result<RecordIdCursor<'_>> {
        let max_records = {
            let table = self.get_table(table_id)?;
            table.def.max_records
        };

        let mut ids_u16 = vec![0u16; max_records];

        let count = {
            let index = self.get_secondary_index_mut(table_id)?;
            unsafe {
                index.find_range_all(
                    start_key,
                    start_key_size,
                    end_key,
                    end_key_size,
                    ids_u16.as_mut_ptr(),
                    ids_u16.len(),
                )?
            }
        };

        let table = self.get_table(table_id)?;
        let ids = ids_u16[..count]
            .iter()
            .map(|id| *id as usize)
            .collect::<Vec<usize>>();
        Ok(table.scan_ids_ref(ids))
    }

    /// 获取表（可变）

    pub fn get_table_mut(&mut self, table_id: usize) -> Result<&mut MemoryTable> {
        if table_id >= self.tables.len() {
            return Err(RemDbError::RecordNotFound);
        }

        match &mut self.tables[table_id] {
            Some(table) => Ok(table),
            None => Err(RemDbError::RecordNotFound),
        }
    }

    /// 获取主键索引
    pub fn get_primary_index(&self, table_id: usize) -> Result<&PrimaryIndex> {
        if table_id >= self.primary_indices.len() {
            return Err(RemDbError::RecordNotFound);
        }

        match &self.primary_indices[table_id] {
            Some(index) => Ok(index),
            None => Err(RemDbError::RecordNotFound),
        }
    }

    /// 获取主键索引（可变）
    pub fn get_primary_index_mut(&mut self, table_id: usize) -> Result<&mut PrimaryIndex> {
        if table_id >= self.primary_indices.len() {
            return Err(RemDbError::RecordNotFound);
        }

        match &mut self.primary_indices[table_id] {
            Some(index) => Ok(index),
            None => Err(RemDbError::RecordNotFound),
        }
    }

    /// 获取辅助索引
    pub fn get_secondary_index(&self, table_id: usize) -> Result<&AnySecondaryIndex> {
        if table_id >= self.secondary_indices.len() {
            return Err(RemDbError::RecordNotFound);
        }

        match &self.secondary_indices[table_id] {
            Some(index) => Ok(index),
            None => Err(RemDbError::RecordNotFound),
        }
    }

    /// 获取辅助索引（可变）
    pub fn get_secondary_index_mut(&mut self, table_id: usize) -> Result<&mut AnySecondaryIndex> {
        if table_id >= self.secondary_indices.len() {
            return Err(RemDbError::RecordNotFound);
        }

        match &mut self.secondary_indices[table_id] {
            Some(index) => Ok(index),
            None => Err(RemDbError::RecordNotFound),
        }
    }

    /// 根据表名获取表引用和辅助索引的可变引用
    pub fn get_table_and_secondary_index_mut_by_name(
        &mut self,
        table_name: &str,
    ) -> Result<(&MemoryTable, &mut AnySecondaryIndex)> {
        // 首先查找表ID
        let mut table_id = None;
        for (id, table_opt) in self.tables.iter().enumerate() {
            if let Some(table) = table_opt {
                if table.def.name == table_name {
                    table_id = Some(id);
                    break;
                }
            }
        }

        let table_id = table_id.ok_or(RemDbError::RecordNotFound)?;

        // 安全地分割借用：分别借用 tables 和 secondary_indices 字段
        // 这是安全的，因为：
        // 1. tables 和 secondary_indices 是不相关的字段
        // 2. 我们使用相同的索引访问这两个字段
        // 3. 没有创建任何别名或悬垂指针
        unsafe {
            let tables_ptr: *const Vec<Option<MemoryTable>> = &self.tables;
            let secondary_indices_ptr: *mut Vec<Option<AnySecondaryIndex>> =
                &mut self.secondary_indices;

            let table = (&(*tables_ptr))
                .get(table_id)
                .and_then(|opt: &Option<MemoryTable>| opt.as_ref())
                .ok_or(RemDbError::RecordNotFound)?;

            let index = (&mut (*secondary_indices_ptr))
                .get_mut(table_id)
                .and_then(|opt: &mut Option<AnySecondaryIndex>| opt.as_mut())
                .ok_or(RemDbError::RecordNotFound)?;

            Ok((table, index))
        }
    }

    /// 检查是否处于低功耗模式
    pub fn is_low_power_mode(&self) -> bool {
        self.low_power_mode
    }

    /// 进入低功耗模式
    pub fn enter_low_power_mode(&mut self) -> Result<()> {
        // 检查配置是否支持低功耗模式
        if !self.config.low_power_mode_supported {
            return Err(RemDbError::UnsupportedOperation);
        }

        // 如果已经处于低功耗模式，直接返回
        if self.low_power_mode {
            return Ok(());
        }

        // 执行低功耗模式准备工作
        // 1. 压缩内存使用：释放不必要的内存
        // 2. 减少索引更新频率
        // 3. 降低事务日志的写入频率

        // 检查当前内存使用情况
        let current_memory = self.config.total_memory;
        if current_memory > self.low_power_memory_limit {
            // 内存使用超出限制，需要进行优化
            // 实现内存优化逻辑
            self.optimize_memory_usage();
        }

        // 设置事务管理器为低功耗模式
        crate::transaction::set_low_power_mode(true);

        // 遍历所有表，设置低功耗模式
        for table in &mut self.tables.iter_mut() {
            if let Some(table) = table {
                table.set_low_power_mode(true, self.config.low_power_max_records);
            }
        }

        // 更新状态
        self.low_power_mode = true;

        // 记录EnterLowPowerMode日志到WAL
        unsafe {
            // 直接使用LogManager写入日志，而不是通过TransactionManager
            let tx_manager = crate::transaction::get_tx_manager();
            if let Some(log_manager) = tx_manager.get_log_manager_mut() {
                // 创建日志项
                let log_item = crate::transaction::LogItem {
                    op_type: crate::transaction::LogOperation::EnterLowPowerMode,
                    table_id: 0, // 低功耗模式是全局操作，不需要特定表ID
                    record_id: 0,
                    old_data_size: 0,
                    new_data_size: 0,
                    tx_id: 0,
                    timestamp: crate::platform::get_timestamp_us(),
                    checksum: 0,
                };

                // 计算校验和
                let calculated_checksum =
                    crate::transaction::Transaction::calculate_log_item_checksum(&log_item);

                let mut final_log_item = log_item;
                final_log_item.checksum = calculated_checksum;

                // 写入日志
                let _ = log_manager.write_log_item(&final_log_item);
                // 立即刷新缓冲区，确保日志被持久化
                let _ = log_manager.flush_buffer();
            }
        }

        Ok(())
    }

    /// 优化内存使用
    fn optimize_memory_usage(&mut self) {
        // 1. 压缩内存使用：释放不必要的内存
        // 2. 减少索引更新频率
        // 3. 降低事务日志的写入频率

        // 遍历所有表，进行内存优化
        for table in &mut self.tables.iter_mut() {
            if let Some(_table) = table {
                // 优化普通表的内存使用
                // 这里可以添加具体的表内存优化逻辑
            }
        }

        // 遍历所有时序表，进行内存优化
        for ts_table in &mut self.time_series_tables.iter_mut() {
            if let Some(_ts_table) = ts_table {
                // 优化时序表的内存使用
                // 这里可以添加具体的时序表内存优化逻辑
            }
        }

        // 降低索引更新频率
        // 这里可以添加索引优化逻辑

        // 降低事务日志的写入频率
        // 这里可以添加事务日志优化逻辑
    }

    /// 退出低功耗模式
    pub fn exit_low_power_mode(&mut self) -> Result<()> {
        // 如果已经不处于低功耗模式，直接返回
        if !self.low_power_mode {
            return Ok(());
        }

        // 执行退出低功耗模式的准备工作
        // 1. 恢复正常的索引更新频率
        // 2. 恢复正常的事务日志写入频率
        // 3. 检查并扩展内存使用（如果需要）

        // 设置事务管理器为正常模式
        crate::transaction::set_low_power_mode(false);

        // 遍历所有表，退出低功耗模式
        for table in &mut self.tables.iter_mut() {
            if let Some(table) = table {
                table.set_low_power_mode(false, None);
            }
        }

        // 更新状态
        self.low_power_mode = false;

        // 记录ExitLowPowerMode日志到WAL
        unsafe {
            // 直接使用LogManager写入日志，而不是通过TransactionManager
            let tx_manager = crate::transaction::get_tx_manager();
            if let Some(log_manager) = tx_manager.get_log_manager_mut() {
                // 创建日志项
                let log_item = crate::transaction::LogItem {
                    op_type: crate::transaction::LogOperation::ExitLowPowerMode,
                    table_id: 0, // 低功耗模式是全局操作，不需要特定表ID
                    record_id: 0,
                    old_data_size: 0,
                    new_data_size: 0,
                    tx_id: 0,
                    timestamp: crate::platform::get_timestamp_us(),
                    checksum: 0,
                };

                // 计算校验和
                let calculated_checksum =
                    crate::transaction::Transaction::calculate_log_item_checksum(&log_item);

                let mut final_log_item = log_item;
                final_log_item.checksum = calculated_checksum;

                // 写入日志
                let _ = log_manager.write_log_item(&final_log_item);
                // 立即刷新缓冲区，确保日志被持久化
                let _ = log_manager.flush_buffer();
            }
        }

        Ok(())
    }

    /// 开始事务
    pub unsafe fn begin_transaction(
        &mut self,
        tx_type: transaction::TransactionType,
        isolation_level: transaction::IsolationLevel,
        tx_buffer: *mut transaction::Transaction,
        log_buffer: *mut transaction::VariableSizeLogItem,
        max_log_items: usize,
    ) -> Result<NonNull<transaction::Transaction>> {
        crate::transaction::begin(
            tx_type,
            isolation_level,
            tx_buffer,
            log_buffer,
            max_log_items,
        )
    }

    /// 提交事务
    pub unsafe fn commit_transaction(&mut self) -> Result<()> {
        crate::transaction::commit()
    }

    /// 回滚事务
    pub unsafe fn rollback_transaction(&mut self) -> Result<()> {
        crate::transaction::rollback()
    }

    /// 刷新WAL日志到磁盘
    pub unsafe fn flush_logs(&mut self) -> Result<()> {
        let tx_manager = crate::transaction::get_tx_manager();
        tx_manager.flush_logs()
    }

    /// 创建检查点
    pub unsafe fn checkpoint(&mut self) -> Result<()> {
        if let Some(log_manager) = crate::transaction::get_log_manager() {
            log_manager.create_checkpoint()
        } else {
            Err(RemDbError::ConfigError)
        }
    }

    /// 初始化数据库
    pub fn init(&mut self) -> Result<()> {
        // 只有当平台尚未初始化时，才使用默认平台
        if crate::platform::PLATFORM.get().is_none() {
            // 默认使用POSIX平台（如果可用）
            #[cfg(feature = "posix")]
            crate::platform::init_platform(crate::platform::posix::get_posix_platform());

            // 如果POSIX平台不可用（例如在Windows上），使用裸机平台作为备选
            #[cfg(not(feature = "posix"))]
            #[cfg(feature = "baremetal")]
            crate::platform::init_platform(crate::platform::baremetal::get_baremetal_platform());
        }

        // 初始化日志管理器（如果配置了日志）
        // 这里使用默认的日志文件路径，实际应用中可以从配置中获取
        #[cfg(feature = "std")]
        {
            // 只有当平台能正常处理文件时，才初始化日志管理器
            // 测试平台的file_open返回null，会导致FileIoError
            use crate::transaction::LogManager;
            use std::path::Path;

            // 构造完整的日志文件路径：log_path目录 + remdb.wal文件名
            let log_dir = self.config.wal_config.log_path;
            let wal_file_path = format!("{}/remdb.wal", log_dir);

            // 确保日志目录存在（仅在std环境下）
            #[cfg(feature = "std")]
            {
                let log_path = Path::new(&log_dir);
                if !log_path.exists() {
                    std::fs::create_dir_all(log_path).unwrap_or(());
                }
            }

            unsafe {
                // 先检查平台是否能正常打开文件且返回有效的句柄
                match crate::platform::file_open(
                    wal_file_path.as_str(),
                    crate::platform::FileMode::ReadWrite,
                ) {
                    Ok(handle) if !handle.is_null() => {
                        // 文件打开成功且句柄有效，关闭并继续初始化日志管理器
                        let _ = crate::platform::file_close(handle);
                        let log_manager = LogManager::new(self.config)?;
                        crate::transaction::set_log_manager(log_manager);
                    }
                    _ => {
                        // 文件打开失败或句柄无效，跳过日志管理器初始化（适用于测试场景）
                    }
                }
            }
        }

        // 初始化系统表
        let tables_created = unsafe { crate::system_tables::init_system_tables(self)? };

        // 只有在创建了新表时才加载RBAC数据
        // 注意：如果系统表已存在，需要加载之前持久化的RBAC数据
        // 如果系统表是新创建的，则使用默认的RBAC数据（已在RbacManager::new()中初始化）
        if !tables_created {
            // 加载RBAC数据从系统表
            // 使用 addr_of_mut! 获取原始指针以避免借用冲突
            // 函数本身是 unsafe 的，调用者负责保证安全
            let manager_ptr = core::ptr::addr_of_mut!(self.rbac_manager);
            let _ = unsafe { RbacManager::load_from_system_tables(&mut *manager_ptr, &*self) };
        }

        // 初始化HA管理器
        #[cfg(feature = "ha")]
        if let Err(_e) = crate::ha::init(self.config) {
            // HA初始化失败，记录错误但不影响数据库主体功能
            // 可以通过日志或监控系统报告
        }

        Ok(())
    }

    /// 保存快照到文件
    pub fn save_snapshot(&mut self, path: &str) -> Result<()> {
        // 打开文件 - 使用Write模式
        let handle = crate::platform::file_open(path, crate::platform::FileMode::Write)
            .map_err(|_| RemDbError::FileIoError)?;

        // 使用defer确保文件关闭
        let _defer = Defer::new(|| {
            let _ = crate::platform::file_close(handle);
        });

        // 增加全局快照版本号
        self.snapshot_version += 1;

        // 写入魔数
        let magic = Self::SNAPSHOT_MAGIC.to_le_bytes();
        let written = crate::platform::file_write(handle, magic.as_ptr(), magic.len())
            .map_err(|_| RemDbError::FileIoError)?;
        if written != magic.len() {
            return Err(RemDbError::FileIoError);
        }

        // 写入版本号
        let version = Self::SNAPSHOT_VERSION.to_le_bytes();
        let written = crate::platform::file_write(handle, version.as_ptr(), version.len())
            .map_err(|_| RemDbError::FileIoError)?;
        if written != version.len() {
            return Err(RemDbError::FileIoError);
        }

        // 写入快照类型：0=完整快照
        let snapshot_type = 0u8;
        let written = crate::platform::file_write(handle, &snapshot_type as *const u8, 1)
            .map_err(|_| RemDbError::FileIoError)?;
        if written != 1 {
            return Err(RemDbError::FileIoError);
        }

        // 写入全局快照版本号
        let version_bytes = self.snapshot_version.to_le_bytes();
        let written =
            crate::platform::file_write(handle, version_bytes.as_ptr(), version_bytes.len())
                .map_err(|_| RemDbError::FileIoError)?;
        if written != version_bytes.len() {
            return Err(RemDbError::FileIoError);
        }

        // 写入表数量（只计算实际存在的表）
        let table_count = self.tables.iter().filter(|t| t.is_some()).count() as u32;
        let table_count_bytes = table_count.to_le_bytes();
        let written = crate::platform::file_write(
            handle,
            table_count_bytes.as_ptr(),
            table_count_bytes.len(),
        )
        .map_err(|_| RemDbError::FileIoError)?;
        if written != table_count_bytes.len() {
            return Err(RemDbError::FileIoError);
        }

        // 写入每个表的数据（只保存实际存在的表）
        for table_id in 0..self.tables.len() {
            if let Some(table) = &mut self.tables[table_id] {
                // 更新表快照版本号
                table.snapshot_version = self.snapshot_version;

                // 写入表ID（4字节）
                let table_id_u32 = table_id as u32;
                let table_id_bytes = table_id_u32.to_le_bytes();
                let written = crate::platform::file_write(
                    handle,
                    table_id_bytes.as_ptr(),
                    table_id_bytes.len(),
                )
                .map_err(|_| RemDbError::FileIoError)?;
                if written != table_id_bytes.len() {
                    return Err(RemDbError::FileIoError);
                }

                // 写入表结构信息
                // 1. 写入表名
                let table_name = &table.def.name;
                let table_name_len = table_name.len() as u8;
                let written = crate::platform::file_write(handle, &table_name_len as *const u8, 1)
                    .map_err(|_| RemDbError::FileIoError)?;
                if written != 1 {
                    return Err(RemDbError::FileIoError);
                }
                let written = crate::platform::file_write(
                    handle,
                    table_name.as_bytes().as_ptr(),
                    table_name.len(),
                )
                .map_err(|_| RemDbError::FileIoError)?;
                if written != table_name.len() {
                    return Err(RemDbError::FileIoError);
                }

                // 2. 写入字段数量
                let field_count = table.def.fields.len() as u8;
                let written = crate::platform::file_write(handle, &field_count as *const u8, 1)
                    .map_err(|_| RemDbError::FileIoError)?;
                if written != 1 {
                    return Err(RemDbError::FileIoError);
                }

                // 3. 写入主键字段数量
                let primary_key_count = table.def.primary_key.len() as u8;
                let written =
                    crate::platform::file_write(handle, &primary_key_count as *const u8, 1)
                        .map_err(|_| RemDbError::FileIoError)?;
                if written != 1 {
                    return Err(RemDbError::FileIoError);
                }

                // 4. 写入辅助索引字段数量
                let secondary_index_count = table
                    .def
                    .secondary_index
                    .as_ref()
                    .map_or(0, |idx| idx.len()) as u8;
                let written =
                    crate::platform::file_write(handle, &secondary_index_count as *const u8, 1)
                        .map_err(|_| RemDbError::FileIoError)?;
                if written != 1 {
                    return Err(RemDbError::FileIoError);
                }

                // 5. 写入辅助索引类型
                let secondary_index_type = table.def.secondary_index_type as u8;
                let written =
                    crate::platform::file_write(handle, &secondary_index_type as *const u8, 1)
                        .map_err(|_| RemDbError::FileIoError)?;
                if written != 1 {
                    return Err(RemDbError::FileIoError);
                }

                // 6. 写入最大记录数
                let max_records_u32 = table.def.max_records as u32;
                let max_records_bytes = max_records_u32.to_le_bytes();
                let written = crate::platform::file_write(
                    handle,
                    max_records_bytes.as_ptr(),
                    max_records_bytes.len(),
                )
                .map_err(|_| RemDbError::FileIoError)?;
                if written != max_records_bytes.len() {
                    return Err(RemDbError::FileIoError);
                }

                // 7. 写入每个字段的完整定义
                for field in &table.def.fields {
                    // 写入字段名称长度和字段名称
                    let field_name_len = field.name.len() as u8;
                    let written =
                        crate::platform::file_write(handle, &field_name_len as *const u8, 1)
                            .map_err(|_| RemDbError::FileIoError)?;
                    if written != 1 {
                        return Err(RemDbError::FileIoError);
                    }
                    let written = crate::platform::file_write(
                        handle,
                        field.name.as_bytes().as_ptr(),
                        field.name.len(),
                    )
                    .map_err(|_| RemDbError::FileIoError)?;
                    if written != field.name.len() {
                        return Err(RemDbError::FileIoError);
                    }

                    // 写入数据类型
                    let data_type = field.data_type as u8;
                    let written = crate::platform::file_write(handle, &data_type as *const u8, 1)
                        .map_err(|_| RemDbError::FileIoError)?;
                    if written != 1 {
                        return Err(RemDbError::FileIoError);
                    }

                    // 写入字段大小
                    let field_size_u32 = field.size as u32;
                    let field_size_bytes = field_size_u32.to_le_bytes();
                    let written = crate::platform::file_write(
                        handle,
                        field_size_bytes.as_ptr(),
                        field_size_bytes.len(),
                    )
                    .map_err(|_| RemDbError::FileIoError)?;
                    if written != field_size_bytes.len() {
                        return Err(RemDbError::FileIoError);
                    }

                    // 写入字符串长度限制（如果有）
                    let has_string_length = field.string_length.is_some() as u8;
                    let written =
                        crate::platform::file_write(handle, &has_string_length as *const u8, 1)
                            .map_err(|_| RemDbError::FileIoError)?;
                    if written != 1 {
                        return Err(RemDbError::FileIoError);
                    }
                    if let Some(len) = field.string_length {
                        let string_len_u32 = len as u32;
                        let string_len_bytes = string_len_u32.to_le_bytes();
                        let written = crate::platform::file_write(
                            handle,
                            string_len_bytes.as_ptr(),
                            string_len_bytes.len(),
                        )
                        .map_err(|_| RemDbError::FileIoError)?;
                        if written != string_len_bytes.len() {
                            return Err(RemDbError::FileIoError);
                        }
                    }

                    // 写入字段标志（主键、非空、唯一、自增）
                    let flags = (field.primary_key as u8) << 0
                        | (field.not_null as u8) << 1
                        | (field.unique as u8) << 2
                        | (field.auto_increment as u8) << 3;
                    let written = crate::platform::file_write(handle, &flags as *const u8, 1)
                        .map_err(|_| RemDbError::FileIoError)?;
                    if written != 1 {
                        return Err(RemDbError::FileIoError);
                    }

                    // 暂时不支持默认值的序列化
                    let has_default = 0u8;
                    let written = crate::platform::file_write(handle, &has_default as *const u8, 1)
                        .map_err(|_| RemDbError::FileIoError)?;
                    if written != 1 {
                        return Err(RemDbError::FileIoError);
                    }
                }

                // 8. 写入主键字段索引列表
                for &pk_idx in &table.def.primary_key {
                    let pk_idx_u8 = pk_idx as u8;
                    let written = crate::platform::file_write(handle, &pk_idx_u8 as *const u8, 1)
                        .map_err(|_| RemDbError::FileIoError)?;
                    if written != 1 {
                        return Err(RemDbError::FileIoError);
                    }
                }

                // 9. 写入辅助索引字段索引列表（如果有）
                if let Some(ref secondary_index) = table.def.secondary_index {
                    for &idx in secondary_index {
                        let idx_u8 = idx as u8;
                        let written = crate::platform::file_write(handle, &idx_u8 as *const u8, 1)
                            .map_err(|_| RemDbError::FileIoError)?;
                        if written != 1 {
                            return Err(RemDbError::FileIoError);
                        }
                    }
                }

                // 10. 写入已使用的记录数（4字节）
                let used_count_u32 = table.record_count() as u32;
                let used_count_bytes = used_count_u32.to_le_bytes();
                let written = crate::platform::file_write(
                    handle,
                    used_count_bytes.as_ptr(),
                    used_count_bytes.len(),
                )
                .map_err(|_| RemDbError::FileIoError)?;
                if written != used_count_bytes.len() {
                    return Err(RemDbError::FileIoError);
                }

                // 动态计算记录大小
                let mut record_size = 0;
                for field in &table.def.fields {
                    record_size += field.size;
                }

                // 写入已使用的记录
                for i in 0..table.def.max_records {
                    let status_ptr = unsafe { table.get_status_ptr(i) };
                    if unsafe { (*status_ptr).status } == crate::types::RecordStatus::Used {
                        // 写入记录索引（4字节）
                        let index_u32 = i as u32;
                        let index_bytes = index_u32.to_le_bytes();
                        let written = crate::platform::file_write(
                            handle,
                            index_bytes.as_ptr(),
                            index_bytes.len(),
                        )
                        .map_err(|_| RemDbError::FileIoError)?;
                        if written != index_bytes.len() {
                            return Err(RemDbError::FileIoError);
                        }

                        // 写入记录数据
                        let record_ptr = unsafe { table.get_record_ptr(i) };
                        let written = crate::platform::file_write(handle, record_ptr, record_size)
                            .map_err(|_| RemDbError::FileIoError)?;
                        if written != record_size {
                            return Err(RemDbError::FileIoError);
                        }
                    }
                }
            }
        }

        // 跳过CRC32计算和写入，简化实现
        Ok(())
    }

    /// 从文件恢复快照
    pub fn restore_snapshot(&mut self, path: &str) -> Result<()> {
        // 打开文件
        let handle = crate::platform::file_open(path, crate::platform::FileMode::Read)
            .map_err(|_| RemDbError::FileIoError)?;

        // 使用defer确保文件关闭
        let _defer = Defer::new(|| {
            let _ = crate::platform::file_close(handle);
        });

        // 读取魔数
        let mut magic_bytes = [0u8; 4];
        let read = crate::platform::file_read(handle, magic_bytes.as_mut_ptr(), magic_bytes.len())
            .map_err(|_| RemDbError::FileIoError)?;
        if read != magic_bytes.len() {
            return Err(RemDbError::FileIoError);
        }
        let magic = u32::from_le_bytes(magic_bytes);
        if magic != Self::SNAPSHOT_MAGIC {
            return Err(RemDbError::SnapshotFormatError);
        }

        // 读取版本号
        let mut version_bytes = [0u8; 4];
        let read =
            crate::platform::file_read(handle, version_bytes.as_mut_ptr(), version_bytes.len())
                .map_err(|_| RemDbError::FileIoError)?;
        if read != version_bytes.len() {
            return Err(RemDbError::FileIoError);
        }
        let version = u32::from_le_bytes(version_bytes);
        if version != Self::SNAPSHOT_VERSION {
            return Err(RemDbError::SnapshotFormatError);
        }

        // 读取快照类型
        let mut snapshot_type_bytes = [0u8; 1];
        let read = crate::platform::file_read(
            handle,
            snapshot_type_bytes.as_mut_ptr(),
            snapshot_type_bytes.len(),
        )
        .map_err(|_| RemDbError::FileIoError)?;
        if read != snapshot_type_bytes.len() {
            return Err(RemDbError::FileIoError);
        }
        let snapshot_type = snapshot_type_bytes[0];

        // 读取基础版本号
        let mut base_version_bytes = [0u8; 4];
        let read = crate::platform::file_read(
            handle,
            base_version_bytes.as_mut_ptr(),
            base_version_bytes.len(),
        )
        .map_err(|_| RemDbError::FileIoError)?;
        if read != base_version_bytes.len() {
            return Err(RemDbError::FileIoError);
        }
        let base_version = u32::from_le_bytes(base_version_bytes);

        // 读取表数量
        let mut table_count_bytes = [0u8; 4];
        let read = crate::platform::file_read(
            handle,
            table_count_bytes.as_mut_ptr(),
            table_count_bytes.len(),
        )
        .map_err(|_| RemDbError::FileIoError)?;
        if read != table_count_bytes.len() {
            return Err(RemDbError::FileIoError);
        }
        let table_count = u32::from_le_bytes(table_count_bytes) as usize;

        // 移除表数量匹配检查，允许从快照恢复表结构

        // 读取每个表的数据
        for _ in 0..table_count {
            // 读取表ID（4字节）
            let mut table_id_bytes = [0u8; 4];
            let read = crate::platform::file_read(
                handle,
                table_id_bytes.as_mut_ptr(),
                table_id_bytes.len(),
            )
            .map_err(|_| RemDbError::FileIoError)?;
            if read != table_id_bytes.len() {
                return Err(RemDbError::FileIoError);
            }
            let table_id = u32::from_le_bytes(table_id_bytes) as usize;

            // 检查并扩展表向量容量
            if table_id >= self.tables.len() {
                self.tables.resize_with(table_id + 1, || None);
                self.primary_indices.resize_with(table_id + 1, || None);
                self.secondary_indices.resize_with(table_id + 1, || None);
            }

            // 获取表引用，如果不存在则跳过
            // 首先读取表名，用于验证表是否匹配
            let mut table_name_len_bytes = [0u8; 1];
            let read = crate::platform::file_read(
                handle,
                table_name_len_bytes.as_mut_ptr(),
                table_name_len_bytes.len(),
            )
            .map_err(|_| RemDbError::FileIoError)?;
            if read != table_name_len_bytes.len() {
                return Err(RemDbError::FileIoError);
            }
            let table_name_len = table_name_len_bytes[0] as usize;

            // 读取表名
            let mut table_name_bytes = vec![0u8; table_name_len];
            let read =
                crate::platform::file_read(handle, table_name_bytes.as_mut_ptr(), table_name_len)
                    .map_err(|_| RemDbError::FileIoError)?;
            if read != table_name_len {
                return Err(RemDbError::FileIoError);
            }
            let table_name = String::from_utf8_lossy(&table_name_bytes).to_string();

            // 读取字段数量
            let mut field_count_bytes = [0u8; 1];
            let read = crate::platform::file_read(
                handle,
                field_count_bytes.as_mut_ptr(),
                field_count_bytes.len(),
            )
            .map_err(|_| RemDbError::FileIoError)?;
            if read != field_count_bytes.len() {
                return Err(RemDbError::FileIoError);
            }
            let field_count = field_count_bytes[0] as usize;

            // 读取主键字段数量
            let mut primary_key_count_bytes = [0u8; 1];
            let read = crate::platform::file_read(
                handle,
                primary_key_count_bytes.as_mut_ptr(),
                primary_key_count_bytes.len(),
            )
            .map_err(|_| RemDbError::FileIoError)?;
            if read != primary_key_count_bytes.len() {
                return Err(RemDbError::FileIoError);
            }
            let primary_key_count = primary_key_count_bytes[0] as usize;

            // 读取辅助索引字段数量
            let mut secondary_index_count_bytes = [0u8; 1];
            let read = crate::platform::file_read(
                handle,
                secondary_index_count_bytes.as_mut_ptr(),
                secondary_index_count_bytes.len(),
            )
            .map_err(|_| RemDbError::FileIoError)?;
            if read != secondary_index_count_bytes.len() {
                return Err(RemDbError::FileIoError);
            }
            let secondary_index_count = secondary_index_count_bytes[0] as usize;

            // 读取辅助索引类型
            let mut secondary_index_type_bytes = [0u8; 1];
            let read = crate::platform::file_read(
                handle,
                secondary_index_type_bytes.as_mut_ptr(),
                secondary_index_type_bytes.len(),
            )
            .map_err(|_| RemDbError::FileIoError)?;
            if read != secondary_index_type_bytes.len() {
                return Err(RemDbError::FileIoError);
            }
            let secondary_index_type = secondary_index_type_bytes[0];

            // 读取最大记录数
            let mut max_records_bytes = [0u8; 4];
            let read = crate::platform::file_read(
                handle,
                max_records_bytes.as_mut_ptr(),
                max_records_bytes.len(),
            )
            .map_err(|_| RemDbError::FileIoError)?;
            if read != max_records_bytes.len() {
                return Err(RemDbError::FileIoError);
            }
            let max_records = u32::from_le_bytes(max_records_bytes) as usize;

            // 读取字段定义
            let mut fields = Vec::new();
            for _ in 0..field_count {
                // 读取字段名称
                let mut field_name_len_bytes = [0u8; 1];
                let read = crate::platform::file_read(
                    handle,
                    field_name_len_bytes.as_mut_ptr(),
                    field_name_len_bytes.len(),
                )
                .map_err(|_| RemDbError::FileIoError)?;
                if read != field_name_len_bytes.len() {
                    return Err(RemDbError::FileIoError);
                }
                let field_name_len = field_name_len_bytes[0] as usize;

                let mut field_name_bytes = vec![0u8; field_name_len];
                let read = crate::platform::file_read(
                    handle,
                    field_name_bytes.as_mut_ptr(),
                    field_name_len,
                )
                .map_err(|_| RemDbError::FileIoError)?;
                if read != field_name_len {
                    return Err(RemDbError::FileIoError);
                }
                let field_name = String::from_utf8_lossy(&field_name_bytes).to_string();

                // 读取数据类型
                let mut data_type_bytes = [0u8; 1];
                let read = crate::platform::file_read(
                    handle,
                    data_type_bytes.as_mut_ptr(),
                    data_type_bytes.len(),
                )
                .map_err(|_| RemDbError::FileIoError)?;
                if read != data_type_bytes.len() {
                    return Err(RemDbError::FileIoError);
                }
                let data_type = crate::types::DataType::from(data_type_bytes[0]);

                // 读取字段大小
                let mut field_size_bytes = [0u8; 4];
                let read = crate::platform::file_read(
                    handle,
                    field_size_bytes.as_mut_ptr(),
                    field_size_bytes.len(),
                )
                .map_err(|_| RemDbError::FileIoError)?;
                if read != field_size_bytes.len() {
                    return Err(RemDbError::FileIoError);
                }
                let field_size = u32::from_le_bytes(field_size_bytes) as usize;

                // 读取字符串长度限制
                let mut has_string_length_bytes = [0u8; 1];
                let read = crate::platform::file_read(
                    handle,
                    has_string_length_bytes.as_mut_ptr(),
                    has_string_length_bytes.len(),
                )
                .map_err(|_| RemDbError::FileIoError)?;
                if read != has_string_length_bytes.len() {
                    return Err(RemDbError::FileIoError);
                }
                let string_length = if has_string_length_bytes[0] == 1 {
                    let mut string_len_bytes = [0u8; 4];
                    let read = crate::platform::file_read(
                        handle,
                        string_len_bytes.as_mut_ptr(),
                        string_len_bytes.len(),
                    )
                    .map_err(|_| RemDbError::FileIoError)?;
                    if read != string_len_bytes.len() {
                        return Err(RemDbError::FileIoError);
                    }
                    Some(u32::from_le_bytes(string_len_bytes) as usize)
                } else {
                    None
                };

                // 读取字段标志
                let mut flags_bytes = [0u8; 1];
                let read =
                    crate::platform::file_read(handle, flags_bytes.as_mut_ptr(), flags_bytes.len())
                        .map_err(|_| RemDbError::FileIoError)?;
                if read != flags_bytes.len() {
                    return Err(RemDbError::FileIoError);
                }
                let flags = flags_bytes[0];
                let primary_key = (flags & 0x01) != 0;
                let not_null = (flags & 0x02) != 0;
                let unique = (flags & 0x04) != 0;
                let auto_increment = (flags & 0x08) != 0;

                // 读取默认值标志（暂时不支持默认值）
                let mut has_default_bytes = [0u8; 1];
                let read = crate::platform::file_read(
                    handle,
                    has_default_bytes.as_mut_ptr(),
                    has_default_bytes.len(),
                )
                .map_err(|_| RemDbError::FileIoError)?;
                if read != has_default_bytes.len() {
                    return Err(RemDbError::FileIoError);
                }

                fields.push(crate::types::FieldDef {
                    name: field_name,
                    data_type,
                    size: field_size,
                    string_length,
                    offset: 0,
                    primary_key,
                    not_null,
                    unique,
                    auto_increment,
                    default_value: None,
                    vector_metadata: None,
                    json_metadata: None,
                });
            }

            // 读取主键字段索引列表
            let mut primary_key_indices = Vec::new();
            for _ in 0..primary_key_count {
                let mut pk_idx_bytes = [0u8; 1];
                let read = crate::platform::file_read(
                    handle,
                    pk_idx_bytes.as_mut_ptr(),
                    pk_idx_bytes.len(),
                )
                .map_err(|_| RemDbError::FileIoError)?;
                if read != pk_idx_bytes.len() {
                    return Err(RemDbError::FileIoError);
                }
                primary_key_indices.push(pk_idx_bytes[0] as usize);
            }

            // 读取辅助索引字段索引列表
            let secondary_index_indices = if secondary_index_count > 0 {
                let mut indices = Vec::new();
                for _ in 0..secondary_index_count {
                    let mut idx_bytes = [0u8; 1];
                    let read =
                        crate::platform::file_read(handle, idx_bytes.as_mut_ptr(), idx_bytes.len())
                            .map_err(|_| RemDbError::FileIoError)?;
                    if read != idx_bytes.len() {
                        return Err(RemDbError::FileIoError);
                    }
                    indices.push(idx_bytes[0] as usize);
                }
                Some(indices)
            } else {
                None
            };

            // 检查表是否存在，如果不存在则创建
            let table = if let Some(table) = &mut self.tables[table_id] {
                table
            } else {
                #[cfg(feature = "log")]
                info!(
                    "Creating table '{}' with ID {} from snapshot...",
                    table_name, table_id
                );

                // 创建表定义
                let table_def = crate::types::TableDef {
                    id: table_id as u8,
                    name: table_name.clone(),
                    fields,
                    primary_key: primary_key_indices,
                    secondary_index: secondary_index_indices,
                    secondary_index_type: crate::types::IndexType::from(secondary_index_type),
                    record_size: 0,
                    max_records,
                    version: 0,
                    created_at: 0,
                    updated_at: 0,
                };

                // 创建表
                let table_def_arc = alloc::sync::Arc::new(table_def);
                let table = crate::table::MemoryTable::new(table_def_arc)
                    .map_err(|_| RemDbError::OutOfMemory)?;

                self.tables[table_id] = Some(table);
                self.tables[table_id]
                    .as_mut()
                    .ok_or(RemDbError::TableNotFound)?
            };

            // 读取记录数
            let mut record_count_bytes = [0u8; 4];
            let read = crate::platform::file_read(
                handle,
                record_count_bytes.as_mut_ptr(),
                record_count_bytes.len(),
            )
            .map_err(|_| RemDbError::FileIoError)?;
            if read != record_count_bytes.len() {
                return Err(RemDbError::FileIoError);
            }
            let record_count = u32::from_le_bytes(record_count_bytes) as usize;

            // 动态计算记录大小
            let mut record_size = 0;
            for field in &table.def.fields {
                record_size += field.size;
            }

            if snapshot_type == 0 {
                // 完整快照：重置所有记录
                for i in 0..table.def.max_records {
                    let status_ptr = unsafe { table.get_status_ptr(i) };
                    let record_ptr = unsafe { table.get_record_ptr_mut(i) };

                    unsafe {
                        (*status_ptr).status = crate::types::RecordStatus::Free;
                        (*status_ptr).version += 1;
                        crate::platform::memset(record_ptr, 0, table.def.record_size);
                    }
                }

                // 重置记录数
                unsafe {
                    table.set_record_count(0);
                }
            }

            // 读取记录数据
            for _ in 0..record_count {
                // 读取记录索引（4字节）
                let mut index_bytes = [0u8; 4];
                let read =
                    crate::platform::file_read(handle, index_bytes.as_mut_ptr(), index_bytes.len())
                        .map_err(|_| RemDbError::FileIoError)?;
                if read != index_bytes.len() {
                    return Err(RemDbError::FileIoError);
                }
                let i = u32::from_le_bytes(index_bytes) as usize;

                // 检查索引是否有效
                if i >= table.def.max_records {
                    return Err(RemDbError::SnapshotFormatError);
                }

                // 读取记录数据
                let record_ptr = unsafe { table.get_record_ptr_mut(i) };
                let read = crate::platform::file_read(handle, record_ptr, record_size)
                    .map_err(|_| RemDbError::FileIoError)?;
                if read != record_size {
                    return Err(RemDbError::FileIoError);
                }

                // 更新记录状态
                let status_ptr = unsafe { table.get_status_ptr(i) };
                let current_status = unsafe { &mut *status_ptr };

                if current_status.status != crate::types::RecordStatus::Used {
                    // 如果记录之前是空闲的，增加记录数
                    unsafe {
                        table.inc_record_count();
                    }
                }

                current_status.status = crate::types::RecordStatus::Used;
                current_status.version += 1;

                // 更新表的max_pk值，确保新插入的记录不会覆盖旧记录
                let record_ptr = unsafe { table.get_record_ptr_mut(i) };
                if table.def.primary_key.len() == 1 {
                    let pk_col_idx = table.def.primary_key[0];
                    let primary_key_field = &table.def.fields[pk_col_idx];
                    let new_pk = unsafe {
                        let key_ptr = record_ptr.add(primary_key_field.offset);
                        match primary_key_field.data_type {
                            crate::types::DataType::UInt8 => {
                                core::ptr::read_unaligned(key_ptr as *const u8) as u64
                            }
                            crate::types::DataType::UInt16 => {
                                core::ptr::read_unaligned(key_ptr as *const u16) as u64
                            }
                            crate::types::DataType::UInt32 => {
                                core::ptr::read_unaligned(key_ptr as *const u32) as u64
                            }
                            crate::types::DataType::UInt64 => {
                                core::ptr::read_unaligned(key_ptr as *const u64)
                            }
                            crate::types::DataType::Int8 => {
                                core::ptr::read_unaligned(key_ptr as *const i8) as u64
                            }
                            crate::types::DataType::Int16 => {
                                core::ptr::read_unaligned(key_ptr as *const i16) as u64
                            }
                            crate::types::DataType::Int32 => {
                                core::ptr::read_unaligned(key_ptr as *const i32) as u64
                            }
                            crate::types::DataType::Int64 => {
                                core::ptr::read_unaligned(key_ptr as *const i64) as u64
                            }
                            _ => 0,
                        }
                    };

                    if new_pk > table.max_pk {
                        table.max_pk = new_pk;
                    }
                }
            }
        }

        // 更新全局快照版本号
        self.snapshot_version = base_version;

        ///// 简化实现，跳过CRC32校验
        Ok(())
    }

    /// 获取当前监控指标
    pub fn get_metrics(&self) -> &monitor::DbMetrics {
        &self.metrics
    }

    /// 创建指标快照
    pub fn metrics_snapshot(&self) -> monitor::DbMetricsSnapshot {
        let snapshot = self.metrics.snapshot();

        // Publish metrics to pubsub
        #[cfg(feature = "pubsub")]
        if let Some(topic_id) = crate::pubsub::get_topic_id(crate::pubsub::topics::METRICS_TOPIC) {
            let metrics_bytes = snapshot.to_bytes();
            let _ = crate::pubsub::publish(topic_id, &metrics_bytes);
        }

        snapshot
    }

    /// 重置所有监控指标
    pub fn reset_metrics(&self) {
        self.metrics.reset()
    }

    /// 执行健康检查
    pub fn health_check(&self) -> monitor::HealthCheckResult {
        let health_result = {
            // 作用域限制，确保snapshot不会被同时引用
            let metrics = self.metrics.snapshot();

            // 健康检查逻辑
            let memory_usage = metrics.used_memory as f64 / metrics.total_memory as f64;

            let (status, details) = if memory_usage > 0.9 {
                (
                    monitor::HealthStatus::Unhealthy,
                    alloc::string::String::from("内存使用率过高"),
                )
            } else if memory_usage > 0.7 {
                (
                    monitor::HealthStatus::Warning,
                    alloc::string::String::from("内存使用率较高"),
                )
            } else {
                (
                    monitor::HealthStatus::Healthy,
                    alloc::string::String::from("数据库运行正常"),
                )
            };

            monitor::HealthCheckResult::new(status, metrics, details)
        };

        // Publish health status to pubsub
        #[cfg(feature = "pubsub")]
        if let Some(topic_id) =
            crate::pubsub::get_topic_id(crate::pubsub::topics::HEALTH_STATUS_TOPIC)
        {
            let health_bytes = health_result.to_bytes();
            let _ = crate::pubsub::publish(topic_id, &health_bytes);
        }

        health_result
    }

    /// 删除表
    ///
    /// # 参数
    /// - `table_name`: 要删除的表名
    /// - `if_exists`: 如果表不存在，是否不报错
    /// - `deferred`: 是否延迟删除
    ///
    /// # 返回值
    /// - `Ok(())`: 删除成功
    /// - `Err(RemDbError)`: 删除失败
    pub fn drop_table(&mut self, table_name: &str, if_exists: bool, _deferred: bool) -> Result<()> {
        #[cfg(feature = "log")]
        info!("Starting DROP TABLE operation on {}", table_name);

        // 输入验证
        if table_name.is_empty() {
            return Err(RemDbError::InvalidArgument);
        }

        // 检查是否为系统表，系统表不允许DROP操作
        if crate::system_tables::is_system_table(table_name) {
            return Err(RemDbError::NotAllowed);
        }

        // 1. 查找表的位置
        let table_index = self.tables.iter().position(|table_opt| {
            if let Some(table) = table_opt {
                table.def.name == table_name
            } else {
                false
            }
        });

        // 2. 处理表不存在的情况
        if table_index.is_none() {
            if if_exists {
                return Ok(());
            } else {
                return Err(RemDbError::TableNotFound);
            }
        }

        let table_index = table_index.ok_or(RemDbError::TableNotFound)?;

        // 防御性检查：确保索引在有效范围内
        if table_index >= self.tables.len() {
            return Err(RemDbError::InvalidState);
        }

        // 增加删除操作计数
        self.metrics.inc_delete_ops();

        // 3. 检查是否有活跃的查询或事务正在使用该表
        // 这里简化处理，实际实现需要更复杂的并发控制

        // 4. 开始事务操作
        if crate::transaction::has_active_tx() {
            // 记录删除操作到事务日志
            unsafe {
                if let Some(mut tx) = crate::transaction::get_current_tx() {
                    let tx_id = tx.as_mut().id;
                    tx.as_mut().begin_log_item(
                        tx_id,
                        crate::transaction::LogOperation::DropTable,
                        table_index as u8,
                        0,
                        0,
                        None,
                        None,
                    );
                }
            }
        }

        // 5. 释放表占用的资源
        let memory_released = if let Some(table) = &self.tables[table_index] {
            // 计算释放的内存大小
            MemoryTable::calculate_memory_size(&table.def)
        } else {
            0
        };

        // 从监控指标中减去释放的内存
        self.metrics.sub_used_memory(memory_released);

        // 移除表（通过将Option设置为None，触发Drop实现）
        self.tables[table_index] = None;

        // 6. 释放对应的索引
        if table_index < self.primary_indices.len() {
            self.primary_indices[table_index] = None;
        }
        if table_index < self.secondary_indices.len() {
            self.secondary_indices[table_index] = None;
        }

        // 7. 从系统表中移除表的条目
        // 这里简化处理，实际实现需要更新系统表

        // 8. 记录删除操作到WAL日志
        unsafe {
            // 直接使用LogManager写入日志，而不是通过TransactionManager
            let tx_manager = crate::transaction::get_tx_manager();
            if let Some(log_manager) = tx_manager.get_log_manager_mut() {
                // 创建日志项
                let mut log_data = [0u8; 512];
                // 写入表名
                let name_bytes = table_name.as_bytes();
                let name_len = core::cmp::min(name_bytes.len(), 64);
                log_data[0] = name_len as u8;
                log_data[1..1 + name_len].copy_from_slice(&name_bytes[..name_len]);

                let log_item = crate::transaction::LogItem {
                    op_type: crate::transaction::LogOperation::DropTable,
                    table_id: table_index as u8,
                    record_id: 0,
                    old_data_size: 0,
                    new_data_size: (name_len + 1) as u16,
                    tx_id: 0,
                    timestamp: crate::platform::get_timestamp_us(),
                    checksum: 0,
                };

                // 计算校验和
                let calculated_checksum =
                    crate::transaction::Transaction::calculate_log_item_checksum(&log_item);

                let mut final_log_item = log_item;
                final_log_item.checksum = calculated_checksum;

                // 写入日志
                if let Err(e) = log_manager.write_log_item(&final_log_item) {
                    #[cfg(feature = "log")]
                    warn!(
                        "Failed to write WAL log for DROP TABLE {}: {:?}",
                        table_name, e
                    );
                    // 继续执行，因为表删除操作已经完成
                }
                // 立即刷新缓冲区，确保日志被持久化
                if let Err(e) = log_manager.flush_buffer() {
                    #[cfg(feature = "log")]
                    warn!(
                        "Failed to flush WAL buffer for DROP TABLE {}: {:?}",
                        table_name, e
                    );
                    // 继续执行，因为表删除操作已经完成
                }
            }
        }

        Ok(())
    }
}

/// 为RemDb实现DdlExecutor trait
impl DdlExecutor for RemDb {
    fn create_table(
        &mut self,
        name: &str,
        fields: &[(&str, DataType, u16, Option<DistanceType>, Option<Value>)],
        constraints: Option<&[FieldConstraint]>,
        primary_key: Option<Vec<usize>>,
        max_records: Option<usize>,
    ) -> Result<()> {
        // 1. 检查字段数量是否合法
        if fields.is_empty() {
            return Err(RemDbError::ConfigError);
        }

        // 2. 检查主键索引是否合法
        if let Some(pk_indices) = &primary_key {
            for &pk_index in pk_indices {
                if pk_index >= fields.len() {
                    return Err(RemDbError::ConfigError);
                }
            }
        }

        // 3. 检查表格是否已存在
        for table_opt in &self.tables {
            if let Some(table) = table_opt {
                if table.def.name == name {
                    // 表格已存在，直接返回成功
                    return Ok(());
                }
            }
        }

        // 3.5. 查找可重用的表ID（优先使用已删除的表ID）
        let table_id = self
            .tables
            .iter()
            .position(|t| t.is_none())
            .unwrap_or(self.tables.len());

        // 3. 计算字段大小和偏移量
        let mut field_defs = Vec::new();
        let mut offset = 0;
        let mut record_size = 0;

        for (i, (field_name, data_type, dimension, distance_type, default_value)) in
            fields.iter().enumerate()
        {
            #[cfg(feature = "log")]
            debug!(
                "field: name={}, type={:?}, dimension={}, distance_type={:?}",
                field_name, data_type, dimension, distance_type
            );
            // 计算字段大小
            let field_size = match data_type {
                DataType::VarChar | DataType::Char => {
                    if *dimension > 0 {
                        *dimension as usize
                    } else {
                        MAX_STRING_LEN
                    }
                }
                DataType::Text => {
                    if *dimension > 0 {
                        *dimension as usize
                    } else {
                        DEFAULT_TEXT_SIZE
                    }
                }
                DataType::Json => {
                    if *dimension > 0 {
                        *dimension as usize
                    } else {
                        DEFAULT_JSON_SIZE
                    }
                }
                DataType::Vector => {
                    // 向量类型：根据压缩配置计算大小
                    crate::system_tables::get_vector_field_size(*dimension)
                }
                _ => data_type.size().unwrap_or(0),
            };

            // 将字段名转换为静态字符串
            let field_name_static = Box::leak(field_name.to_string().into_boxed_str());

            // 检查是否为自增主键
            let is_primary_key = primary_key
                .as_ref()
                .map(|pk| pk.contains(&i))
                .unwrap_or(false);
            // 获取字段约束信息
            let default_constraint = FieldConstraint {
                primary_key: is_primary_key,
                not_null: is_primary_key,
                unique: is_primary_key,
                auto_increment: is_primary_key, // 主键默认支持自增
            };
            let constraint = constraints
                .and_then(|c| c.get(i))
                .unwrap_or(&default_constraint);
            let is_auto_increment = constraint.auto_increment
                && (data_type == &DataType::Int32
                    || data_type == &DataType::Int64
                    || data_type == &DataType::UInt32
                    || data_type == &DataType::UInt64);

            // 主键必须是非空的，覆盖用户设置
            let final_not_null = is_primary_key || constraint.not_null;

            // 主键必须是唯一的，覆盖用户设置
            let final_unique = is_primary_key || constraint.unique;

            // 获取当前向量压缩配置
            let compression_config = crate::system_tables::get_vector_compression_config();

            // 创建向量元数据（仅向量类型需要）
            let vector_metadata = if *data_type == DataType::Vector {
                Some(VectorMetadata {
                    dimension: *dimension,
                    distance_type: distance_type.unwrap_or(DistanceType::L2),
                    index_type: VectorIndexType::HNSW, // 默认使用HNSW索引
                    compression_enabled: compression_config.vector_compression_enabled,
                    compression_scheme: compression_config.vector_compression_scheme as u8,
                    compression_level: compression_config.vector_compression_level,
                    hnsw_m: 16,
                    hnsw_ef_construction: 200,
                    hnsw_ef_search: 128,
                    ivf_nlist: 1024,
                    ivf_nprobe: 16,
                })
            } else {
                None
            };

            // 创建字段定义，设置默认约束
            let field_def = FieldDef {
                name: field_name_static.to_string(),
                data_type: *data_type,
                size: field_size,
                string_length: None, // 暂时设置为None，后续从SQL解析中获取
                offset,
                primary_key: is_primary_key,       // 主键索引匹配当前字段
                not_null: final_not_null,          // 应用非空约束
                unique: final_unique,              // 应用唯一约束
                auto_increment: is_auto_increment, // 应用自增约束
                default_value: default_value.clone(), // 设置字段默认值
                vector_metadata,                   // 设置向量元数据
                json_metadata: None,               // JSON元数据
            };

            field_defs.push(field_def);

            // 更新偏移量和记录大小
            offset += field_size;
            record_size += field_size;
        }

        // 4. 计算有效的max_records
        // 如果低功耗模式开启，实际的最大记录数以max_records与低功耗模式下的最大记录数的较小值为准
        // 如果低功耗模式关闭，实际的最大记录数以max_records与默认最大记录数的较大值为准
        let effective_max_records = if self.config.low_power_mode_supported {
            let low_power_max = self.config.low_power_max_records.unwrap_or(usize::MAX);
            match max_records {
                Some(table_max) => table_max.min(low_power_max),
                None => self.config.default_max_records.min(low_power_max),
            }
        } else {
            match max_records {
                Some(table_max) => table_max.max(self.config.default_max_records),
                None => self.config.default_max_records,
            }
        };

        // 5. 创建表定义
        let now = crate::platform::get_timestamp_us();
        let table_def = TableDef {
            id: table_id as u8,
            name: name.to_string(),
            fields: field_defs,
            primary_key: primary_key.unwrap_or_default(),
            secondary_index: None,
            secondary_index_type: IndexType::SortedArray,
            record_size,
            max_records: effective_max_records,
            version: 1,
            created_at: now,
            updated_at: now,
        };

        // 6. 创建内存表
        let table_def_arc = alloc::sync::Arc::new(table_def.clone());
        let table = MemoryTable::new(table_def_arc.clone())?;

        // 7. 添加到表向量（在可重用位置或末尾）
        if table_id < self.tables.len() {
            self.tables[table_id] = Some(table);
        } else {
            self.tables.push(Some(table));
        }

        // 8. 创建主键索引
        // 计算主键索引所需内存大小
        let hash_table_size = (table_def.max_records * 2).next_power_of_two(); // 哈希表大小为记录数的2倍，取最近的2的幂
        let index_memory_size =
            PrimaryIndex::calculate_memory_size(&table_def, hash_table_size, table_def.max_records);

        // 分配内存
        let index_memory = crate::memory::allocator::alloc(index_memory_size)?;
        let hash_table_start = index_memory.as_ptr() as *mut Option<NonNull<PrimaryIndexItem>>;
        let items_start = (index_memory.as_ptr() as usize
            + hash_table_size * core::mem::size_of::<Option<NonNull<PrimaryIndexItem>>>())
            as *mut PrimaryIndexItem;

        // 创建主键索引
        let primary_index = unsafe {
            PrimaryIndex::new(
                table_def_arc.clone(),
                hash_table_start,
                items_start,
                hash_table_size,
                table_def.max_records,
            )
        };
        if table_id < self.primary_indices.len() {
            self.primary_indices[table_id] = Some(primary_index);
        } else {
            self.primary_indices.push(Some(primary_index));
        }

        // 9. 初始化辅助索引位置
        if table_id < self.secondary_indices.len() {
            self.secondary_indices[table_id] = None;
        } else {
            self.secondary_indices.push(None);
        }

        // Publish table creation to pubsub
        #[cfg(feature = "pubsub")]
        let table_creation_msg = alloc::format!(
            "CREATE:table={},id={},fields={}",
            table_def.name,
            table_def.id,
            table_def.fields.len()
        );

        #[cfg(feature = "pubsub")]
        if let Some(topic_id) = crate::pubsub::get_topic_id(crate::pubsub::topics::TABLES_TOPIC) {
            let _ = crate::pubsub::publish(topic_id, table_creation_msg.as_bytes());
        }

        // 记录CREATE_TABLE日志到WAL
        unsafe {
            // 直接使用LogManager写入日志，而不是通过TransactionManager
            let tx_manager = crate::transaction::get_tx_manager();
            if let Some(log_manager) = tx_manager.get_log_manager_mut() {
                // 序列化表定义信息
                let mut log_data = [0u8; 512];
                // 写入表名
                let name_bytes = table_def.name.as_bytes();
                let name_len = core::cmp::min(name_bytes.len(), 64);
                log_data[0] = name_len as u8;
                log_data[1..1 + name_len].copy_from_slice(&name_bytes[..name_len]);
                // 写入字段数量
                log_data[65] = table_def.fields.len() as u8;
                // 写入主键字段数量
                log_data[66] = table_def.primary_key.len() as u8;
                // 写入主键索引列表
                for (i, &pk_col) in table_def.primary_key.iter().enumerate() {
                    if i + 67 < log_data.len() {
                        log_data[67 + i] = pk_col as u8;
                    }
                }

                // 写入字段定义信息
                let mut offset = 67 + table_def.primary_key.len();
                for (_i, field) in table_def.fields.iter().enumerate() {
                    // 检查缓冲区是否有足够空间写入基础字段信息
                    // 基础信息：1字节长度 + 32字节名字 + 1字节类型 + 1字节约束 + 1字节默认值标志 + 2字节向量维度 = 38字节
                    if offset + 38 > log_data.len() {
                        break;
                    }

                    // 写入字段名
                    let field_name = field.name.clone();
                    let field_name_bytes = field_name.as_bytes();
                    let field_name_len = core::cmp::min(field_name_bytes.len(), 32);

                    // 安全写入字段名长度
                    log_data[offset] = field_name_len as u8;
                    offset += 1;

                    // 安全复制字段名
                    let copy_end = core::cmp::min(offset + field_name_len, log_data.len());
                    let actual_copy_len = copy_end - offset;
                    log_data[offset..copy_end]
                        .copy_from_slice(&field_name_bytes[..actual_copy_len]);
                    // 固定32字节字段名空间，但要确保不超过缓冲区边界
                    offset = core::cmp::min(offset + 32, log_data.len());

                    // 检查数据类型写入边界
                    if offset < log_data.len() {
                        // 写入数据类型
                        log_data[offset] = field.data_type as u8;
                        offset += 1;
                    } else {
                        break;
                    }

                    // 写入字段约束
                    if offset < log_data.len() {
                        let mut constraints = 0u8;
                        if field.primary_key {
                            constraints |= 0b0001;
                        }
                        if field.not_null {
                            constraints |= 0b0010;
                        }
                        if field.unique {
                            constraints |= 0b0100;
                        }
                        if field.auto_increment {
                            constraints |= 0b1000;
                        }
                        log_data[offset] = constraints;
                        offset += 1;
                    } else {
                        break;
                    }

                    // 写入字段大小（4字节）
                    if offset + 4 <= log_data.len() {
                        let field_size_u32 = field.size as u32;
                        log_data[offset..offset + 4].copy_from_slice(&field_size_u32.to_le_bytes());
                        offset += 4;
                    } else {
                        break;
                    }

                    // 写入向量维度（如果是向量类型）
                    if offset + 2 <= log_data.len() {
                        let mut vector_dimension = 0u16;
                        if field.data_type == crate::types::DataType::Vector {
                            if let Some(metadata) = &field.vector_metadata {
                                vector_dimension = metadata.dimension;
                            }
                        }
                        log_data[offset..offset + 2]
                            .copy_from_slice(&vector_dimension.to_le_bytes());
                        offset += 2;
                    } else {
                        break;
                    }

                    // 写入默认值存在标志
                    if offset < log_data.len() {
                        let has_default = field.default_value.is_some();
                        log_data[offset] = has_default as u8;
                        offset += 1;
                    } else {
                        break;
                    }

                    // 写入默认值（如果有）
                    if let Some(default_value) = &field.default_value {
                        // 根据数据类型写入默认值，添加完善的边界检查
                        match field.data_type {
                            // 向量类型
                            crate::types::DataType::Vector => {
                                // 向量默认值处理
                                if let Some(metadata) = &field.vector_metadata {
                                    let vector_size = metadata.dimension as usize * 4; // float32
                                    if offset + vector_size <= log_data.len() {
                                        let vector_ptr = default_value.vector;
                                        std::ptr::copy(
                                            vector_ptr as *const u8,
                                            log_data.as_mut_ptr().add(offset),
                                            vector_size,
                                        );
                                        offset += vector_size;
                                    }
                                }
                            }
                            // 1字节类型
                            crate::types::DataType::Bool
                            | crate::types::DataType::Int8
                            | crate::types::DataType::UInt8 => {
                                if offset + 1 <= log_data.len() {
                                    match field.data_type {
                                        crate::types::DataType::Bool => {
                                            log_data[offset] = default_value.bool as u8;
                                        }
                                        crate::types::DataType::Int8 => {
                                            log_data[offset] = default_value.i8 as u8;
                                        }
                                        _ => {
                                            log_data[offset] = default_value.u8;
                                        }
                                    }
                                    offset += 1;
                                }
                            }
                            // 2字节类型
                            crate::types::DataType::Int16 | crate::types::DataType::UInt16 => {
                                if offset + 2 <= log_data.len() {
                                    let bytes = match field.data_type {
                                        crate::types::DataType::Int16 => {
                                            default_value.i16.to_le_bytes()
                                        }
                                        _ => default_value.u16.to_le_bytes(),
                                    };
                                    log_data[offset..offset + 2].copy_from_slice(&bytes);
                                    offset += 2;
                                }
                            }
                            // 4字节类型
                            crate::types::DataType::Int32
                            | crate::types::DataType::UInt32
                            | crate::types::DataType::Float32 => {
                                if offset + 4 <= log_data.len() {
                                    let bytes = match field.data_type {
                                        crate::types::DataType::Int32 => {
                                            default_value.i32.to_le_bytes()
                                        }
                                        crate::types::DataType::UInt32 => {
                                            default_value.u32.to_le_bytes()
                                        }
                                        _ => default_value.float32.to_le_bytes(),
                                    };
                                    log_data[offset..offset + 4].copy_from_slice(&bytes);
                                    offset += 4;
                                }
                            }
                            // 8字节类型
                            crate::types::DataType::Int64
                            | crate::types::DataType::UInt64
                            | crate::types::DataType::Float64
                            | crate::types::DataType::Timestamp
                            | crate::types::DataType::TimestampTZ => {
                                if offset + 8 <= log_data.len() {
                                    let bytes = match field.data_type {
                                        crate::types::DataType::Int64 => {
                                            default_value.i64.to_le_bytes()
                                        }
                                        crate::types::DataType::UInt64 => {
                                            default_value.u64.to_le_bytes()
                                        }
                                        crate::types::DataType::Float64 => {
                                            default_value.float64.to_le_bytes()
                                        }
                                        _ => default_value.time.value.to_le_bytes(),
                                    };
                                    log_data[offset..offset + 8].copy_from_slice(&bytes);
                                    offset += 8;
                                }
                            }
                            // 字符串类型：1字节长度 + 64字节内容
                            crate::types::DataType::VarChar
                            | crate::types::DataType::Char
                            | crate::types::DataType::Text => {
                                if offset + 65 <= log_data.len() {
                                    let s = default_value.string;
                                    let string_len = core::cmp::min(
                                        s.iter().position(|&c| c == 0).unwrap_or(64),
                                        64,
                                    );
                                    log_data[offset] = string_len as u8;
                                    offset += 1;

                                    // 安全复制字符串内容
                                    let str_end =
                                        core::cmp::min(offset + string_len, log_data.len());
                                    let actual_str_len = str_end - offset;
                                    log_data[offset..str_end].copy_from_slice(&s[..actual_str_len]);
                                    // 固定64字节字符串空间，但要确保不超过缓冲区边界
                                    offset = core::cmp::min(offset + 64, log_data.len());
                                }
                            }
                            // 区间类型：8字节值 + 1字节精度 + 1字节标志 = 10字节
                            crate::types::DataType::Interval => {
                                if offset + 10 <= log_data.len() {
                                    log_data[offset..offset + 8].copy_from_slice(
                                        &default_value.interval.value.to_le_bytes(),
                                    );
                                    offset += 8;
                                    log_data[offset] = default_value.interval.precision;
                                    offset += 1;
                                    log_data[offset] = default_value.interval.flags;
                                    offset += 1;
                                }
                            }
                            // JSON类型
                            crate::types::DataType::Json => {
                                // JSON默认值处理
                                if offset + 64 <= log_data.len() {
                                    // JsonStorage大小
                                    let json_storage = default_value.json_storage;
                                    let storage_ptr = &json_storage as *const _ as *const u8;
                                    std::ptr::copy(
                                        storage_ptr,
                                        log_data.as_mut_ptr().add(offset),
                                        core::mem::size_of::<crate::types::JsonStorage>(),
                                    );
                                    offset += core::mem::size_of::<crate::types::JsonStorage>();
                                }
                            }
                        }
                    }
                }

                let new_data_size = log_data.len() as u16;

                let mut var_log_item = crate::transaction::VariableSizeLogItem {
                    header: crate::transaction::LogItem {
                        op_type: crate::transaction::LogOperation::CreateTable,
                        table_id: table_def.id,
                        record_id: 0,
                        old_data_size: 0,
                        new_data_size,
                        tx_id: 0,
                        timestamp: crate::platform::get_timestamp_us(),
                        checksum: 0,
                    },
                    old_data: Vec::new(),
                    new_data: log_data.to_vec(),
                };

                let calculated_checksum =
                    crate::transaction::Transaction::calculate_variable_size_log_item_checksum(
                        &var_log_item,
                    );

                var_log_item.header.checksum = calculated_checksum;

                let _ = log_manager.write_variable_size_log_item(&var_log_item);
                let _ = log_manager.flush_buffer();
            }
        }

        Ok(())
    }

    fn create_index(
        &mut self,
        table_name: &str,
        field_name: &str,
        index_type: IndexType,
    ) -> Result<()> {
        // 1. 查找表
        let table_id = self
            .tables
            .iter()
            .position(|t| {
                if let Some(table) = t {
                    table.def.name == table_name
                } else {
                    false
                }
            })
            .ok_or(RemDbError::TableNotFound)?;

        // 2. 查找字段
        let table = self.tables[table_id]
            .as_ref()
            .ok_or(RemDbError::TableNotFound)?;
        let field_index = table
            .def
            .fields
            .iter()
            .position(|f| f.name == field_name)
            .ok_or(RemDbError::FieldNotFound)?;

        // 3. 对于向量索引，检查向量维度是否有效
        let field = &table.def.fields[field_index];
        if index_type == IndexType::Vector {
            // 向量索引必须使用向量类型的字段
            if field.data_type != DataType::Vector {
                #[cfg(feature = "log")]
                error!("TypeMismatch in create_index: field.data_type != DataType::Vector, actual: {:?}, field: {:?}", field.data_type, field.name);
                return Err(RemDbError::TypeMismatch);
            }

            let vector_meta = match field.vector_metadata.as_ref() {
                Some(meta) => meta,
                None => {
                    #[cfg(feature = "log")]
                    error!(
                        "TypeMismatch in create_index: field.vector_metadata is None, field: {:?}",
                        field.name
                    );
                    return Err(RemDbError::TypeMismatch);
                }
            };
            // 索引的向量列维度必须在1-1024范围内
            if vector_meta.dimension == 0 || vector_meta.dimension > 1024 {
                #[cfg(feature = "log")]
                error!(
                    "TypeMismatch in create_index: invalid dimension: {}, field: {:?}",
                    vector_meta.dimension, field.name
                );
                return Err(RemDbError::TypeMismatch);
            }
        }

        // 4. 检查是否已存在索引
        // 确保secondary_indices向量有足够的容量
        while self.secondary_indices.len() <= table_id {
            self.secondary_indices.push(None);
        }
        if self.secondary_indices[table_id].is_some() {
            // 索引已存在，直接返回成功（幂等操作）
            return Ok(());
        }

        // 6. 创建一个新的Arc<TableDef>，包含索引信息
        let new_def = alloc::sync::Arc::new(TableDef {
            id: table.def.id,
            name: table.def.name.clone(),
            fields: table.def.fields.clone(),
            primary_key: table.def.primary_key.clone(),
            secondary_index: Some(vec![field_index]),
            secondary_index_type: index_type,
            record_size: table.def.record_size,
            max_records: table.def.max_records,
            version: table.def.version,
            created_at: table.def.created_at,
            updated_at: table.def.updated_at,
        });

        // 7. 为索引分配内存
        let max_items = table.def.max_records;

        // 对于BTree和TTree索引，减少节点数量，避免占用过多内存导致测试卡住
        let index_max_nodes = match index_type {
            IndexType::BTree | IndexType::TTree => 100, // 只使用100个节点的容量
            IndexType::SortedArray => max_items,        // 有序数组索引使用原始值
            IndexType::Hash => max_items,               // 哈希索引使用原始值
            IndexType::Vector => max_items,             // 向量索引使用原始值
            IndexType::Json => max_items,               // JSON索引使用原始值
        };

        // 计算索引所需内存大小
        let index_size =
            AnySecondaryIndex::calculate_memory_size(new_def.as_ref(), index_max_nodes);

        // 为索引分配内存
        let index_memory = crate::memory::allocator::alloc(index_size)?;

        // 获取索引统计信息
        let stats = crate::memory::allocator::get_memory_stats();
        #[cfg(feature = "log")]
        debug!(
            "Index creation stats: index_size={}, used={}, total={}",
            index_size, stats.used, stats.total
        );

        // 创建索引
        let index =
            unsafe { AnySecondaryIndex::new(new_def, index_memory.as_ptr(), index_max_nodes)? };

        // 存储索引
        self.secondary_indices[table_id] = Some(index);

        // 9. 记录CREATE_INDEX日志到WAL
        unsafe {
            let tx_manager = crate::transaction::get_tx_manager();
            if let Some(log_manager) = tx_manager.get_log_manager_mut() {
                let mut log_data = alloc::vec::Vec::new();
                let table_name_bytes = table_name.as_bytes();
                let table_name_len = core::cmp::min(table_name_bytes.len(), 64);
                log_data.push(table_name_len as u8);
                log_data.extend_from_slice(&table_name_bytes[..table_name_len]);
                let field_name_bytes = field_name.as_bytes();
                let field_name_len = core::cmp::min(field_name_bytes.len(), 64);
                log_data.resize(66, 0);
                log_data[65] = field_name_len as u8;
                log_data.extend_from_slice(&field_name_bytes[..field_name_len]);
                log_data.resize(130, 0);
                log_data.push(index_type as u8);

                let new_data_size = log_data.len() as u16;

                let mut var_log_item = crate::transaction::VariableSizeLogItem {
                    header: crate::transaction::LogItem {
                        op_type: crate::transaction::LogOperation::CreateIndex,
                        table_id: table.def.id,
                        record_id: 0,
                        old_data_size: 0,
                        new_data_size,
                        tx_id: 0,
                        timestamp: crate::platform::get_timestamp_us(),
                        checksum: 0,
                    },
                    old_data: Vec::new(),
                    new_data: log_data,
                };

                let calculated_checksum =
                    crate::transaction::Transaction::calculate_variable_size_log_item_checksum(
                        &var_log_item,
                    );

                var_log_item.header.checksum = calculated_checksum;

                let _ = log_manager.write_variable_size_log_item(&var_log_item);
            }
        }

        Ok(())
    }

    fn create_time_series_table(
        &mut self,
        name: &str,
        time_field: &str,
        value_field: &str,
        tag_fields: &[&str],
        config: Option<TimeSeriesConfig>,
    ) -> Result<()> {
        // 调用RemDb结构体的create_time_series_table方法
        RemDb::create_time_series_table(self, name, time_field, value_field, tag_fields, config)
    }

    fn alter_table(&mut self, table_name: &str, operation: AlterTableOperation) -> Result<()> {
        #[cfg(feature = "log")]
        info!("Starting ALTER TABLE operation on {}", table_name);

        // 检查是否为系统表，系统表不允许ALTER操作
        if crate::system_tables::is_system_table(table_name) {
            return Err(RemDbError::NotAllowed);
        }

        // 1. 查找表
        let table_index = self
            .tables
            .iter()
            .position(|table_opt| {
                if let Some(table) = table_opt {
                    table.def.name == table_name
                } else {
                    false
                }
            })
            .ok_or(RemDbError::TableNotFound)?;

        #[cfg(feature = "log")]
        debug!("Found table at index {}", table_index);

        // 2. 获取当前表定义
        let current_table = self.tables[table_index]
            .as_ref()
            .ok_or(RemDbError::TableNotFound)?;
        let mut new_table_def = (*current_table.def).clone();

        #[cfg(feature = "log")]
        debug!("Current table def: {:?}", current_table.def.name);

        // 3. 根据操作类型执行相应的表结构变更
        #[cfg(feature = "log")]
        debug!("Executing operation: {:?}", operation);

        match operation {
            AlterTableOperation::AddColumn {
                ref name,
                data_type,
                size,
                distance_type,
                ref default_value,
                ref constraints,
            } => {
                // 检查列名是否已存在
                if new_table_def.fields.iter().any(|f| f.name == *name) {
                    return Err(RemDbError::ConfigError);
                }

                // 计算新字段的偏移量
                let new_offset = new_table_def.record_size;

                // 计算字段大小
                let field_size = match data_type {
                    DataType::VarChar | DataType::Char | DataType::Text => size as usize,
                    DataType::Vector => {
                        // 向量类型：维度 * 4字节（f32）
                        size as usize * 4
                    }
                    _ => data_type.size().unwrap_or(0),
                };

                // 解析约束条件
                let primary_key = constraints.primary_key;
                let not_null = constraints.not_null;
                let unique = constraints.unique;
                let auto_increment = constraints.auto_increment;

                // 创建新字段定义
                let new_field = FieldDef {
                    name: name.clone(),
                    data_type,
                    size: field_size,
                    string_length: if matches!(data_type, DataType::VarChar | DataType::Char) {
                        Some(size as usize)
                    } else {
                        None
                    },
                    offset: new_offset,
                    primary_key,
                    not_null,
                    unique,
                    auto_increment,
                    default_value: default_value.clone(),
                    vector_metadata: if data_type == DataType::Vector {
                        Some(VectorMetadata {
                            dimension: size,
                            distance_type: distance_type.unwrap_or(DistanceType::L2),
                            index_type: VectorIndexType::HNSW,
                            compression_enabled: false,
                            compression_scheme: 0,
                            compression_level: 3,
                            hnsw_m: 16,
                            hnsw_ef_construction: 200,
                            hnsw_ef_search: 128,
                            ivf_nlist: 1024,
                            ivf_nprobe: 16,
                        })
                    } else {
                        None
                    },
                    json_metadata: None,
                };

                // 更新表定义
                new_table_def.fields.push(new_field);
                new_table_def.record_size += field_size;
                new_table_def.version += 1;
                new_table_def.updated_at = crate::platform::get_timestamp_us();
            }
            AlterTableOperation::DropColumn { ref name } => {
                // 查找要删除的列
                let field_index = new_table_def
                    .fields
                    .iter()
                    .position(|f| f.name == *name)
                    .ok_or(RemDbError::FieldNotFound)?;

                // 不能删除主键列
                if new_table_def.fields[field_index].primary_key {
                    return Err(RemDbError::ConfigError);
                }

                // 获取要删除字段的大小
                let _field_size = new_table_def.fields[field_index].size;

                // 删除字段
                new_table_def.fields.remove(field_index);

                // 更新剩余字段的偏移量
                let mut new_offset = 0;
                for field in &mut new_table_def.fields {
                    field.offset = new_offset;
                    new_offset += field.size;
                }

                // 更新记录大小
                new_table_def.record_size = new_offset;
                new_table_def.version += 1;
                new_table_def.updated_at = crate::platform::get_timestamp_us();
            }
            AlterTableOperation::ModifyColumn {
                ref name,
                data_type,
                size,
                distance_type,
                ref default_value,
                ref constraints,
            } => {
                // 查找要修改的列
                let field_index = new_table_def
                    .fields
                    .iter()
                    .position(|f| f.name == *name)
                    .ok_or(RemDbError::FieldNotFound)?;

                let field = &mut new_table_def.fields[field_index];
                let old_size = field.size;

                // 计算新字段大小
                let new_size = match data_type {
                    DataType::VarChar | DataType::Char | DataType::Text => size as usize,
                    DataType::Vector => {
                        // 向量类型：维度 * 4字节（f32）
                        size as usize * 4
                    }
                    _ => data_type.size().unwrap_or(0),
                };

                // 更新字段定义
                field.data_type = data_type;
                field.size = new_size;
                field.string_length = if matches!(data_type, DataType::VarChar | DataType::Char) {
                    Some(size as usize)
                } else {
                    None
                };
                field.default_value = default_value.clone();
                field.primary_key = constraints.primary_key;
                field.not_null = constraints.not_null;
                field.unique = constraints.unique;
                field.auto_increment = constraints.auto_increment;

                // 更新向量元数据
                if data_type == DataType::Vector {
                    field.vector_metadata = Some(VectorMetadata {
                        dimension: size,
                        distance_type: distance_type.unwrap_or(DistanceType::L2),
                        index_type: VectorIndexType::HNSW,
                        compression_enabled: false,
                        compression_scheme: 0,
                        compression_level: 3,
                        hnsw_m: 16,
                        hnsw_ef_construction: 200,
                        hnsw_ef_search: 128,
                        ivf_nlist: 1024,
                        ivf_nprobe: 16,
                    });
                } else {
                    field.vector_metadata = None;
                }

                // 如果字段大小改变，需要重新计算所有后续字段的偏移量
                if new_size != old_size {
                    let size_diff = new_size - old_size;

                    // 更新后续字段的偏移量
                    for i in field_index + 1..new_table_def.fields.len() {
                        new_table_def.fields[i].offset += size_diff;
                    }

                    // 更新记录大小
                    new_table_def.record_size += size_diff;
                }

                new_table_def.version += 1;
                new_table_def.updated_at = crate::platform::get_timestamp_us();
            }
            AlterTableOperation::RenameColumn {
                ref old_name,
                ref new_name,
            } => {
                // 检查新列名是否已存在
                if new_table_def.fields.iter().any(|f| f.name == *new_name) {
                    return Err(RemDbError::ConfigError);
                }

                // 查找要重命名的列
                let field_index = new_table_def
                    .fields
                    .iter()
                    .position(|f| f.name == *old_name)
                    .ok_or(RemDbError::FieldNotFound)?;

                // 重命名列
                new_table_def.fields[field_index].name = new_name.clone();
                new_table_def.version += 1;
                new_table_def.updated_at = crate::platform::get_timestamp_us();
            }
        }

        // 4. 计算主键索引所需内存大小
        #[cfg(feature = "log")]
        debug!("Calculating primary index memory size");

        let hash_table_size = (new_table_def.max_records * 2).next_power_of_two(); // 哈希表大小为记录数的2倍，取最近的2的幂
        let index_memory_size = PrimaryIndex::calculate_memory_size(
            &new_table_def,
            hash_table_size,
            new_table_def.max_records,
        );

        #[cfg(feature = "log")]
        debug!(
            "Hash table size: {}, index memory size: {}",
            hash_table_size, index_memory_size
        );

        // 分配内存
        #[cfg(feature = "log")]
        debug!("Allocating index memory");

        let index_memory = crate::memory::allocator::alloc(index_memory_size).map_err(|e| {
            #[cfg(feature = "log")]
            error!("Failed to allocate index memory: {:?}", e);
            e
        })?;

        #[cfg(feature = "log")]
        debug!("Allocated index memory at {:?}", index_memory.as_ptr());

        let hash_table_start = index_memory.as_ptr() as *mut Option<NonNull<PrimaryIndexItem>>;
        let items_start = (index_memory.as_ptr() as usize
            + hash_table_size * core::mem::size_of::<Option<NonNull<PrimaryIndexItem>>>())
            as *mut PrimaryIndexItem;

        // 5. 创建新的内存表
        #[cfg(feature = "log")]
        debug!("Creating new table with updated definition");

        let new_table_def_arc = alloc::sync::Arc::new(new_table_def);
        let mut new_table = MemoryTable::new(new_table_def_arc.clone()).map_err(|e| {
            #[cfg(feature = "log")]
            error!("Failed to create new table: {:?}", e);
            e
        })?;

        #[cfg(feature = "log")]
        debug!("Created new table successfully");

        // 6. 迁移旧表数据到新表并同时构建索引
        #[cfg(feature = "log")]
        debug!("Starting data migration and index construction");

        // 保存旧表引用
        let old_table = current_table;

        // 创建新的主键索引
        let mut primary_index = unsafe {
            PrimaryIndex::new(
                new_table_def_arc.clone(),
                hash_table_start,
                items_start,
                hash_table_size,
                new_table_def_arc.max_records,
            )
        };

        // 迁移数据
        unsafe {
            // 遍历旧表的所有记录槽
            for slot_id in 0..old_table.def.max_records {
                let status_ptr = old_table.status_array.as_ptr().add(slot_id);
                let status = &*status_ptr;

                // 只迁移已使用且可见的记录
                if status.status == RecordStatus::Used {
                    // 获取旧记录的数据指针
                    let old_record_ptr = old_table
                        .data_start
                        .as_ptr()
                        .add(slot_id * old_table.record_size);

                    // 创建新记录缓冲区，大小为新表的记录大小
                    let mut new_record_data = vec![0u8; new_table.record_size];

                    // 迁移字段数据
                    match &operation {
                        AlterTableOperation::RenameColumn { .. } => {
                            // For RenameColumn, match fields by position since only the name changes
                            for (i, old_field) in old_table.def.fields.iter().enumerate() {
                                if i < new_table_def_arc.fields.len() {
                                    let new_field = &new_table_def_arc.fields[i];
                                    // 复制字段数据
                                    let copy_len = core::cmp::min(old_field.size, new_field.size);
                                    crate::platform::memcpy(
                                        new_record_data.as_mut_ptr().add(new_field.offset),
                                        old_record_ptr.add(old_field.offset),
                                        copy_len,
                                    );
                                }
                            }
                        }
                        _ => {
                            // For other operations, match fields by name
                            for old_field in old_table.def.fields.iter() {
                                // 查找新表中对应的字段（按名称匹配）
                                if let Some(new_field) = new_table_def_arc
                                    .fields
                                    .iter()
                                    .find(|f| f.name == old_field.name)
                                {
                                    // 复制字段数据
                                    let copy_len = core::cmp::min(old_field.size, new_field.size);
                                    crate::platform::memcpy(
                                        new_record_data.as_mut_ptr().add(new_field.offset),
                                        old_record_ptr.add(old_field.offset),
                                        copy_len,
                                    );
                                }
                            }

                            // 为新添加的字段设置默认值
                            for new_field in new_table_def_arc.fields.iter() {
                                // 检查是否为新字段（旧表中不存在）
                                if !old_table
                                    .def
                                    .fields
                                    .iter()
                                    .any(|f| f.name == new_field.name)
                                {
                                    // 如果有默认值，设置默认值
                                    if let Some(default_value) = &new_field.default_value {
                                        unsafe {
                                            match new_field.data_type {
                                                DataType::UInt8 => {
                                                    new_record_data[new_field.offset] =
                                                        default_value.u8;
                                                }
                                                DataType::UInt16 => {
                                                    core::ptr::write_unaligned(
                                                        new_record_data
                                                            .as_mut_ptr()
                                                            .add(new_field.offset)
                                                            as *mut u16,
                                                        default_value.u16,
                                                    );
                                                }
                                                DataType::UInt32 => {
                                                    core::ptr::write_unaligned(
                                                        new_record_data
                                                            .as_mut_ptr()
                                                            .add(new_field.offset)
                                                            as *mut u32,
                                                        default_value.u32,
                                                    );
                                                }
                                                DataType::UInt64 => {
                                                    core::ptr::write_unaligned(
                                                        new_record_data
                                                            .as_mut_ptr()
                                                            .add(new_field.offset)
                                                            as *mut u64,
                                                        default_value.u64,
                                                    );
                                                }
                                                DataType::Int8 => {
                                                    new_record_data[new_field.offset] =
                                                        default_value.i8 as u8;
                                                }
                                                DataType::Int16 => {
                                                    core::ptr::write_unaligned(
                                                        new_record_data
                                                            .as_mut_ptr()
                                                            .add(new_field.offset)
                                                            as *mut i16,
                                                        default_value.i16,
                                                    );
                                                }
                                                DataType::Int32 => {
                                                    core::ptr::write_unaligned(
                                                        new_record_data
                                                            .as_mut_ptr()
                                                            .add(new_field.offset)
                                                            as *mut i32,
                                                        default_value.i32,
                                                    );
                                                }
                                                DataType::Int64 => {
                                                    core::ptr::write_unaligned(
                                                        new_record_data
                                                            .as_mut_ptr()
                                                            .add(new_field.offset)
                                                            as *mut i64,
                                                        default_value.i64,
                                                    );
                                                }
                                                DataType::Float32 => {
                                                    core::ptr::write_unaligned(
                                                        new_record_data
                                                            .as_mut_ptr()
                                                            .add(new_field.offset)
                                                            as *mut f32,
                                                        default_value.float32,
                                                    );
                                                }
                                                DataType::Float64 => {
                                                    core::ptr::write_unaligned(
                                                        new_record_data
                                                            .as_mut_ptr()
                                                            .add(new_field.offset)
                                                            as *mut f64,
                                                        default_value.float64,
                                                    );
                                                }
                                                DataType::Bool => {
                                                    new_record_data[new_field.offset] =
                                                        if default_value.bool { 1u8 } else { 0u8 };
                                                }
                                                DataType::Timestamp => {
                                                    core::ptr::write_unaligned(
                                                        new_record_data
                                                            .as_mut_ptr()
                                                            .add(new_field.offset)
                                                            as *mut crate::types::db_timestamp,
                                                        default_value.time,
                                                    );
                                                }
                                                DataType::TimestampTZ => {
                                                    core::ptr::write_unaligned(
                                                        new_record_data
                                                            .as_mut_ptr()
                                                            .add(new_field.offset)
                                                            as *mut crate::types::db_timestamp,
                                                        default_value.time,
                                                    );
                                                }
                                                DataType::VarChar
                                                | DataType::Char
                                                | DataType::Text
                                                | DataType::Json => {
                                                    // 字符串类型：直接写入字符串数据
                                                    let bytes = &default_value.string;
                                                    let len = core::cmp::min(
                                                        MAX_STRING_LEN,
                                                        new_field.size,
                                                    );
                                                    unsafe {
                                                        core::ptr::copy_nonoverlapping(
                                                            bytes.as_ptr(),
                                                            new_record_data
                                                                .as_mut_ptr()
                                                                .add(new_field.offset),
                                                            len,
                                                        );
                                                    }
                                                }
                                                _ => {
                                                    // 其他类型暂不处理
                                                }
                                            }
                                        }
                                    }
                                }
                            }
                        }
                    }

                    // 直接插入新记录到新表，绕过约束验证和事务处理
                    // 获取空闲槽
                    let new_slot_id;
                    let _is_overwrite = false;

                    // 自旋锁保护
                    crate::platform::spin_lock(&mut new_table.lock);

                    // 检查是否有空闲槽
                    if new_table.free_slot_count > 0 {
                        // 从空闲槽栈获取空闲记录槽
                        new_slot_id = *new_table
                            .free_slots
                            .as_ptr()
                            .add(new_table.free_slot_count - 1);
                        new_table.free_slot_count -= 1;
                    } else {
                        // 没有空闲槽，跳过这条记录
                        crate::platform::spin_unlock(&mut new_table.lock);
                        continue;
                    }

                    // 释放锁
                    crate::platform::spin_unlock(&mut new_table.lock);

                    // 计算记录地址
                    let record_ptr = new_table
                        .data_start
                        .as_ptr()
                        .add(new_slot_id * new_table.record_size);

                    // 拷贝记录数据
                    crate::platform::memcpy(
                        record_ptr,
                        new_record_data.as_ptr(),
                        new_table.record_size,
                    );

                    // 更新状态
                    let status_ptr = new_table.status_array.as_ptr().add(new_slot_id);
                    (*status_ptr).status = crate::types::RecordStatus::Used;
                    (*status_ptr).version += 1;

                    // 再次加锁，更新记录计数和max_pk
                    crate::platform::spin_lock(&mut new_table.lock);
                    new_table.record_count += 1;

                    // 更新max_pk（如果有主键字段）
                    if let Some(pk_field) = new_table_def_arc.fields.iter().find(|f| f.primary_key)
                    {
                        // 获取当前记录的主键值
                        let pk_value = match pk_field.data_type {
                            crate::types::DataType::UInt8 => core::ptr::read_unaligned(
                                record_ptr.add(pk_field.offset) as *const u8,
                            ) as u64,
                            crate::types::DataType::UInt16 => core::ptr::read_unaligned(
                                record_ptr.add(pk_field.offset) as *const u16,
                            ) as u64,
                            crate::types::DataType::UInt32 => core::ptr::read_unaligned(
                                record_ptr.add(pk_field.offset) as *const u32,
                            ) as u64,
                            crate::types::DataType::UInt64 => core::ptr::read_unaligned(
                                record_ptr.add(pk_field.offset) as *const u64,
                            ),
                            crate::types::DataType::Int8 => core::ptr::read_unaligned(
                                record_ptr.add(pk_field.offset) as *const i8,
                            ) as u64,
                            crate::types::DataType::Int16 => core::ptr::read_unaligned(
                                record_ptr.add(pk_field.offset) as *const i16,
                            ) as u64,
                            crate::types::DataType::Int32 => core::ptr::read_unaligned(
                                record_ptr.add(pk_field.offset) as *const i32,
                            ) as u64,
                            crate::types::DataType::Int64 => core::ptr::read_unaligned(
                                record_ptr.add(pk_field.offset) as *const i64,
                            ) as u64,
                            _ => 0,
                        };

                        if pk_value > new_table.max_pk {
                            new_table.max_pk = pk_value;
                        }
                    }

                    crate::platform::spin_unlock(&mut new_table.lock);

                    // 插入到主键索引（不需要锁，因为索引还未被使用）
                    if !new_table_def_arc.primary_key.is_empty() {
                        primary_index
                            .insert_composite(record_ptr, new_slot_id as u16)
                            .map_err(|e| {
                                #[cfg(feature = "log")]
                                error!("Failed to insert into primary index: {:?}", e);
                                e
                            })?;
                    }
                }
            }
        }

        #[cfg(feature = "log")]
        debug!("Data migration completed successfully");

        // 7. 替换旧表
        self.tables[table_index] = Some(new_table);

        // 8. 替换旧的主键索引
        self.primary_indices[table_index] = Some(primary_index);

        // 10. 重置辅助索引（因为表结构已经改变）
        self.secondary_indices[table_index] = None;

        // 10. 记录ALTER_TABLE日志到WAL
        unsafe {
            // 直接使用LogManager写入日志，而不是通过TransactionManager
            let tx_manager = crate::transaction::get_tx_manager();
            if let Some(log_manager) = tx_manager.get_log_manager_mut() {
                // 序列化表结构变更信息
                let mut log_data = [0u8; 512];

                // 写入表名
                let name_bytes = table_name.as_bytes();
                let name_len = core::cmp::min(name_bytes.len(), 64);
                log_data[0] = name_len as u8;
                log_data[1..1 + name_len].copy_from_slice(&name_bytes[..name_len]);

                // 写入操作类型
                let op_type_code = match &operation {
                    AlterTableOperation::AddColumn { .. } => 0,
                    AlterTableOperation::DropColumn { .. } => 1,
                    AlterTableOperation::ModifyColumn { .. } => 2,
                    AlterTableOperation::RenameColumn { .. } => 3,
                };
                log_data[65] = op_type_code;

                // 写入表ID
                log_data[66] = table_index as u8;

                // 根据操作类型写入详细信息
                let mut data_size = 67;
                match &operation {
                    AlterTableOperation::AddColumn {
                        ref name,
                        ref data_type,
                        size,
                        distance_type: _,
                        default_value: _,
                        ref constraints,
                    } => {
                        // 写入列名
                        let name_bytes = name.as_bytes();
                        let col_name_len = core::cmp::min(name_bytes.len(), 64);
                        log_data[67] = col_name_len as u8;
                        log_data[68..68 + col_name_len]
                            .copy_from_slice(&name_bytes[..col_name_len]);

                        // 写入数据类型
                        log_data[132] = (*data_type) as u8;

                        // 写入大小
                        log_data[133..135].copy_from_slice(&size.to_le_bytes());

                        // 写入约束
                        let mut constraint_bits = 0u8;
                        if constraints.primary_key {
                            constraint_bits |= 0b0001;
                        }
                        if constraints.not_null {
                            constraint_bits |= 0b0010;
                        }
                        if constraints.unique {
                            constraint_bits |= 0b0100;
                        }
                        if constraints.auto_increment {
                            constraint_bits |= 0b1000;
                        }
                        log_data[135] = constraint_bits;

                        data_size = 136;
                    }
                    AlterTableOperation::DropColumn { ref name } => {
                        // 写入列名
                        let name_bytes = name.as_bytes();
                        let col_name_len = core::cmp::min(name_bytes.len(), 64);
                        log_data[67] = col_name_len as u8;
                        log_data[68..68 + col_name_len]
                            .copy_from_slice(&name_bytes[..col_name_len]);

                        data_size = 68 + col_name_len;
                    }
                    AlterTableOperation::ModifyColumn {
                        ref name,
                        ref data_type,
                        size,
                        distance_type: _,
                        default_value: _,
                        ref constraints,
                    } => {
                        // 写入列名
                        let name_bytes = name.as_bytes();
                        let col_name_len = core::cmp::min(name_bytes.len(), 64);
                        log_data[67] = col_name_len as u8;
                        log_data[68..68 + col_name_len]
                            .copy_from_slice(&name_bytes[..col_name_len]);

                        // 写入数据类型
                        log_data[132] = (*data_type) as u8;

                        // 写入大小
                        log_data[133..135].copy_from_slice(&size.to_le_bytes());

                        // 写入约束
                        let mut constraint_bits = 0u8;
                        if constraints.primary_key {
                            constraint_bits |= 0b0001;
                        }
                        if constraints.not_null {
                            constraint_bits |= 0b0010;
                        }
                        if constraints.unique {
                            constraint_bits |= 0b0100;
                        }
                        if constraints.auto_increment {
                            constraint_bits |= 0b1000;
                        }
                        log_data[135] = constraint_bits;

                        data_size = 136;
                    }
                    AlterTableOperation::RenameColumn {
                        ref old_name,
                        ref new_name,
                    } => {
                        // 写入旧列名
                        let old_name_bytes = old_name.as_bytes();
                        let old_col_name_len = core::cmp::min(old_name_bytes.len(), 64);
                        log_data[67] = old_col_name_len as u8;
                        log_data[68..68 + old_col_name_len]
                            .copy_from_slice(&old_name_bytes[..old_col_name_len]);

                        // 写入新列名
                        let new_name_bytes = new_name.as_bytes();
                        let new_col_name_len = core::cmp::min(new_name_bytes.len(), 64);
                        log_data[132] = new_col_name_len as u8;
                        log_data[133..133 + new_col_name_len]
                            .copy_from_slice(&new_name_bytes[..new_col_name_len]);

                        data_size = 133 + new_col_name_len;
                    }
                }

                // 创建日志项
                let log_item = crate::transaction::LogItem {
                    op_type: crate::transaction::LogOperation::AlterTable,
                    table_id: table_index as u8,
                    record_id: 0, // ALTER TABLE操作不涉及特定记录
                    old_data_size: 0,
                    new_data_size: data_size as u16,
                    tx_id: 0,
                    timestamp: crate::platform::get_timestamp_us(),
                    checksum: 0,
                };

                // 计算校验和
                let calculated_checksum =
                    crate::transaction::Transaction::calculate_log_item_checksum(&log_item);

                let mut final_log_item = log_item;
                final_log_item.checksum = calculated_checksum;

                // 写入日志
                let _ = log_manager.write_log_item(&final_log_item);
                // 立即刷新缓冲区，确保日志被持久化
                let _ = log_manager.flush_buffer();
            }
        }

        Ok(())
    }
}

impl RemDb {
    /// 将指标输出为文本格式
    pub fn dump_metrics(&self) -> alloc::string::String {
        self.metrics.snapshot().to_text()
    }

    /// 执行SQL查询
    pub fn sql_query(&mut self, sql: &str) -> Result<sql::ResultSet> {
        #[cfg(feature = "log")]
        debug!("Executing SQL: {}", sql);

        // 解析SQL查询
        let query = crate::sql::parse_sql_query(sql).map_err(|e| {
            #[cfg(feature = "log")]
            error!("SQL Parse Error: {:?}", e);
            RemDbError::InvalidSqlQuery
        })?;

        // 执行查询
        let result_set = crate::sql::execute_query(self, &query).map_err(|err| {
            #[cfg(feature = "log")]
            error!("SQL Execution Error: {:?}", err);
            match err {
                crate::sql::QueryExecutionError::TableNotFound => RemDbError::TableNotFound,
                crate::sql::QueryExecutionError::FieldNotFound => RemDbError::FieldNotFound,
                crate::sql::QueryExecutionError::TypeMismatch => RemDbError::TypeMismatch,
                crate::sql::QueryExecutionError::ConstraintsConflicts => RemDbError::DuplicateKey,
                crate::sql::QueryExecutionError::OutOfMemory => RemDbError::OutOfMemory,
                crate::sql::QueryExecutionError::NotAllowed => RemDbError::NotAllowed,
                crate::sql::QueryExecutionError::DatabaseExists => RemDbError::DatabaseExists,
                _ => {
                    #[cfg(feature = "log")]
                    error!("Unhandled execution error: {:?}", err);
                    RemDbError::InternalError
                }
            }
        })?;

        Ok(result_set)
    }

    /// 执行查询操作
    pub fn execute_query(
        &mut self,
        table_name: &str,
        columns: &[&str],
        where_clause: Option<&str>,
        limit: Option<usize>,
    ) -> Result<sql::ResultSet> {
        // 构建SELECT SQL语句
        let select_columns = if columns.is_empty() {
            "*".to_string() // 返回String类型
        } else {
            columns.join(", ") // 返回String类型
        };

        let mut sql = alloc::format!("SELECT {} FROM {}", select_columns, table_name);

        if let Some(where_clause) = where_clause {
            sql.push_str(&alloc::format!(" WHERE {}", where_clause));
        }

        if let Some(limit) = limit {
            sql.push_str(&alloc::format!(" LIMIT {}", limit));
        }

        // 调用sql_query执行
        self.sql_query(&sql)
    }

    /// 创建表
    pub fn create_table(
        &mut self,
        table_name: &str,
        fields: &[(&str, DataType, u16, Option<DistanceType>, Option<Value>)],
        primary_key: Option<Vec<usize>>,
    ) -> Result<()> {
        // 调用已有的DdlExecutor实现，不传递约束信息
        // 注意：这个方法签名不包含约束参数，所以无法传递约束信息
        // 如果需要支持约束，应该使用 DdlExecutor::create_table 直接调用
        DdlExecutor::create_table(self, table_name, fields, None, primary_key, None)
    }

    /// 创建表（带约束支持）
    pub fn create_table_with_constraints(
        &mut self,
        table_name: &str,
        fields: &[(&str, DataType, u16, Option<DistanceType>, Option<Value>)],
        constraints: Option<&[FieldConstraint]>,
        primary_key: Option<Vec<usize>>,
        max_records: Option<usize>,
    ) -> Result<()> {
        DdlExecutor::create_table(
            self,
            table_name,
            fields,
            constraints,
            primary_key,
            max_records,
        )
    }

    /// 创建时序表
    pub fn create_time_series_table(
        &mut self,
        name: &str,
        time_field: &str,
        value_field: &str,
        tag_fields: &[&str],
        config: Option<TimeSeriesConfig>,
    ) -> Result<()> {
        // 1. 准备字段定义
        // 时序表至少包含时间字段、值字段和标签字段
        let mut field_defs = Vec::new();
        let mut offset = 0;
        let mut record_size = 0;

        // 添加时间字段（TIMESTAMP）
        let time_field_name = time_field.to_string();
        let time_field_size = DataType::Timestamp.size_unwrap();
        field_defs.push(FieldDef {
            name: time_field_name.clone(),
            data_type: DataType::Timestamp,
            size: time_field_size,
            string_length: None,
            offset,
            primary_key: true,
            not_null: true,
            unique: false,
            auto_increment: false,
            default_value: Some(Value {
                time: crate::types::db_timestamp::new(0, 0, 0, 0),
            }),
            vector_metadata: None,
            json_metadata: None,
        });
        offset += time_field_size;
        record_size += time_field_size;

        // 添加值字段（FLOAT64）
        let value_field_name = value_field.to_string();
        let value_field_size = DataType::Float64.size_unwrap();
        field_defs.push(FieldDef {
            name: value_field_name.clone(),
            data_type: DataType::Float64,
            size: value_field_size,
            string_length: None,
            offset,
            primary_key: false,
            not_null: true,
            unique: false,
            auto_increment: false,
            default_value: Some(Value { float64: 0.0 }),
            vector_metadata: None,
            json_metadata: None,
        });
        offset += value_field_size;
        record_size += value_field_size;

        // 添加标签字段（VARCHAR）
        let mut tag_field_indices = Vec::new();
        for (i, tag_field) in tag_fields.iter().enumerate() {
            let tag_field_name = tag_field.to_string();
            let tag_field_size = MAX_STRING_LEN; // VARCHAR使用最大字符串长度
            field_defs.push(FieldDef {
                name: tag_field_name.clone(),
                data_type: DataType::VarChar,
                size: tag_field_size,
                string_length: None,
                offset,
                primary_key: false,
                not_null: false,
                unique: false,
                auto_increment: false,
                default_value: None, // 标签字段默认值为None
                vector_metadata: None,
                json_metadata: None,
            });
            tag_field_indices.push((i + 2) as usize); // 时间字段(0) + 值字段(1) + 标签字段(i)
            offset += tag_field_size;
            record_size += tag_field_size;
        }

        // 2. 创建表定义
        let table_def = TableDef {
            id: (self.tables.len() + self.time_series_tables.len()) as u8,
            name: name.to_string(),
            fields: field_defs,
            primary_key: vec![0], // 时间字段作为主键
            secondary_index: None,
            secondary_index_type: IndexType::SortedArray,
            record_size,
            max_records: self.config.default_max_records,
            version: 1,
            created_at: crate::platform::get_timestamp_us(),
            updated_at: crate::platform::get_timestamp_us(),
        };

        // 3. 创建时序表定义
        let time_series_table_def = time_series::TimeSeriesTableDef {
            base: table_def,
            time_field: 0,                                    // 时间字段索引
            value_field: 1,                                   // 值字段索引
            tag_fields: tag_field_indices.into_boxed_slice(), // 标签字段索引列表
            config: config.unwrap_or(time_series::TimeSeriesConfig::DEFAULT), // 时序数据配置
        };

        // 4. 创建时序索引
        let index = time_series::TimeSeriesIndex::new();

        // 5. 创建时序表
        let time_series_table =
            time_series::TimeSeriesTable::new(Arc::new(time_series_table_def), index)?;

        // 6. 添加到时序表向量
        self.time_series_tables.push(Some(time_series_table));

        Ok(())
    }

    /// 获取时序表
    pub fn get_time_series_table(&self, table_id: usize) -> Result<&time_series::TimeSeriesTable> {
        if table_id >= self.time_series_tables.len() {
            return Err(RemDbError::RecordNotFound);
        }

        match &self.time_series_tables[table_id] {
            Some(table) => Ok(table),
            None => Err(RemDbError::RecordNotFound),
        }
    }

    /// 获取时序表（可变）
    pub fn get_time_series_table_mut(
        &mut self,
        table_id: usize,
    ) -> Result<&mut time_series::TimeSeriesTable> {
        if table_id >= self.time_series_tables.len() {
            return Err(RemDbError::RecordNotFound);
        }

        match &mut self.time_series_tables[table_id] {
            Some(table) => Ok(table),
            None => Err(RemDbError::RecordNotFound),
        }
    }

    /// 获取表数量
    pub fn table_count(&self) -> usize {
        self.tables.len()
    }

    /// 获取时序表数量
    pub fn time_series_table_count(&self) -> usize {
        self.time_series_tables.len()
    }

    /// 获取所有表的引用
    pub fn get_all_tables(&self) -> &Vec<Option<MemoryTable>> {
        &self.tables
    }

    /// 事务化批量写入时序数据
    /// 确保一批数据要么全部成功插入并立即可见，要么全部回滚
    pub fn write_timeseries_batch(
        &mut self,
        table_name: &str,
        data_points: &[time_series::TimeSeriesRecord],
    ) -> Result<usize> {
        if data_points.is_empty() {
            return Err(RemDbError::ConfigError);
        }

        // 查找时序表
        let table_id = self
            .time_series_tables
            .iter()
            .position(|table| {
                if let Some(table) = table {
                    table.def.base.name == table_name
                } else {
                    false
                }
            })
            .ok_or(RemDbError::TableNotFound)?;

        let table = match &mut self.time_series_tables[table_id] {
            Some(table) => table,
            None => return Err(RemDbError::TableNotFound),
        };

        // 执行批量写入
        // 注意：当前实现依赖于外部事务管理，或者使用内部自动事务
        // 为了简化实现，我们直接执行批量写入，不尝试创建事务

        // 检查是否有活跃事务
        let has_active_tx = crate::transaction::has_active_tx();

        // 如果没有活跃事务，我们使用一个简化的事务管理方式
        // 直接执行写入操作，确保原子性
        let mut inserted = 0;

        if !has_active_tx {
            // 没有活跃事务，直接执行批量写入，不记录日志
            // 注意：这不是完整的ACID事务，但确保了基本的批量写入功能
            for record in data_points {
                // 获取或创建分区
                let mut partitions_guard = try_lock!(table.partitions);
                let partition = partitions_guard.get_or_create_partition(record.timestamp);

                // 写入记录到分区
                let mut partition_guard = try_lock!(partition);
                partition_guard.records.push(*record);
                partition_guard.stats.record_count = partition_guard.records.len();

                // 更新索引
                table.index.insert(record.timestamp, inserted as usize);

                inserted += 1;
            }

            Ok(inserted)
        } else {
            // 已有活跃事务，执行批量写入并记录日志
            table.write_timeseries_batch(data_points)
        }
    }

    /// 创建索引
    pub fn create_index(
        &mut self,
        table_name: &str,
        field_name: &str,
        index_type: IndexType,
    ) -> Result<()> {
        // 调用已有的DdlExecutor实现
        DdlExecutor::create_index(self, table_name, field_name, index_type)
    }

    /// 插入记录
    pub fn insert_record(
        &mut self,
        table_name: &str,
        column_names: &[&str],
        values: &[&str],
    ) -> Result<usize> {
        // 构建INSERT SQL语句
        let columns = if column_names.is_empty() {
            "".to_string() // 返回String类型
        } else {
            alloc::format!(
                "({})
",
                column_names.join(", ")
            ) // 返回String类型
        };

        // 处理值，为字符串值添加引号
        let quoted_values: Vec<String> = values
            .iter()
            .map(|&value| {
                // 检查是否是数值类型或布尔值
                if value
                    .chars()
                    .all(|c| c.is_digit(10) || c == '.' || c == '-')
                    || value == "true"
                    || value == "false"
                {
                    value.to_string()
                } else {
                    // 字符串类型，添加引号
                    alloc::format!("'{}'", value)
                }
            })
            .collect();

        let values_str = alloc::format!(
            "({})
",
            quoted_values.join(", ")
        );

        let sql = alloc::format!(
            "INSERT INTO {}{} VALUES {}",
            table_name,
            columns,
            values_str
        );

        // 执行查询
        let result_set = self.sql_query(&sql)?;

        // 从结果集中获取受影响的行数
        if let Some(row) = result_set.rows.first() {
            if let Some(value) = row.values.first() {
                // 假设第一个值是受影响的行数（u64类型）
                unsafe {
                    let affected_rows = value.value.u64 as usize;
                    return Ok(affected_rows);
                }
            }
        }

        Ok(0)
    }

    /// 批量插入记录
    pub fn batch_insert_record(
        &mut self,
        table_name: &str,
        column_names: &[&str],
        records: &[&[&str]],
    ) -> Result<usize> {
        if records.is_empty() {
            return Ok(0);
        }

        // 构建INSERT SQL语句
        let columns = if column_names.is_empty() {
            "".to_string()
        } else {
            alloc::format!(
                "({})
",
                column_names.join(", ")
            )
        };

        // 处理所有记录的值，为字符串值添加引号
        let mut all_values: Vec<String> = Vec::with_capacity(records.len());

        for values in records {
            let quoted_values: Vec<String> = values
                .iter()
                .map(|&value| {
                    // 检查是否是数值类型或布尔值
                    if value
                        .chars()
                        .all(|c| c.is_digit(10) || c == '.' || c == '-')
                        || value == "true"
                        || value == "false"
                    {
                        value.to_string()
                    } else {
                        // 字符串类型，添加引号
                        alloc::format!("'{}'", value)
                    }
                })
                .collect();

            all_values.push(alloc::format!(
                "({})
",
                quoted_values.join(", ")
            ));
        }

        let values_str = all_values.join(", ");
        let sql = alloc::format!(
            "INSERT INTO {}{} VALUES {}",
            table_name,
            columns,
            values_str
        );

        // 执行查询
        let result_set = self.sql_query(&sql)?;

        // 从结果集中获取受影响的行数
        if let Some(row) = result_set.rows.first() {
            if let Some(value) = row.values.first() {
                // 假设第一个值是受影响的行数（u64类型）
                unsafe {
                    let affected_rows = value.value.u64 as usize;
                    return Ok(affected_rows);
                }
            }
        }

        Ok(0)
    }

    /// 更新记录
    pub fn update_record(
        &mut self,
        table_name: &str,
        set_clause: &str,
        where_clause: Option<&str>,
    ) -> Result<usize> {
        // 构建UPDATE SQL语句
        let mut sql = alloc::format!("UPDATE {} SET {}", table_name, set_clause);

        if let Some(where_clause) = where_clause {
            sql.push_str(&alloc::format!(" WHERE {}", where_clause));
        }

        // 执行查询
        let result_set = self.sql_query(&sql)?;

        // 从结果集中获取受影响的行数
        if let Some(row) = result_set.rows.first() {
            if let Some(value) = row.values.first() {
                // 假设第一个值是受影响的行数（u64类型）
                unsafe {
                    let affected_rows = value.value.u64 as usize;
                    return Ok(affected_rows);
                }
            }
        }

        Ok(0)
    }

    /// 删除记录
    pub fn delete_record(&mut self, table_name: &str, where_clause: Option<&str>) -> Result<usize> {
        // 构建DELETE SQL语句
        let mut sql = alloc::format!("DELETE FROM {}", table_name);

        if let Some(where_clause) = where_clause {
            sql.push_str(&alloc::format!(" WHERE {}", where_clause));
        }

        // 执行查询
        let result_set = self.sql_query(&sql)?;

        // 从结果集中获取受影响的行数
        if let Some(row) = result_set.rows.first() {
            if let Some(value) = row.values.first() {
                // 假设第一个值是受影响的行数（u64类型）
                unsafe {
                    let affected_rows = value.value.u64 as usize;
                    return Ok(affected_rows);
                }
            }
        }

        Ok(0)
    }

    /// 导出完整的DDL文件
    #[cfg(feature = "std")]
    pub fn export_ddl(&self, path: &str) -> Result<()> {
        // 使用标准库的文件操作
        use std::fs::File;
        use std::io::Write;

        let mut file = File::create(path).map_err(|_| RemDbError::FileIoError)?;

        // 遍历所有普通表
        for table_id in 0..self.tables.len() {
            if let Some(table) = &self.tables[table_id] {
                // 生成CREATE TABLE语句，表名转换为小写
                let mut create_table_sql = alloc::string::String::new();
                create_table_sql.push_str(&format!(
                    "CREATE TABLE {} (\n",
                    table.def.name.to_lowercase()
                ));

                // 添加字段定义
                let mut fields_sql = Vec::new();
                for field in &table.def.fields {
                    let field_sql = format!(
                        "    {} {} {}",
                        field.name,
                        field.data_type.to_sql_type(field.size),
                        field.constraints_to_sql()
                    );
                    fields_sql.push(field_sql);
                }

                // 连接字段定义
                create_table_sql.push_str(&fields_sql.join(",\n"));
                create_table_sql.push_str("\n);\n\n");

                // 写入CREATE TABLE语句
                file.write_all(create_table_sql.as_bytes())
                    .map_err(|_| RemDbError::FileIoError)?;

                // 生成CREATE INDEX语句（如果有辅助索引）
                if let Some(secondary_index) = &table.def.secondary_index {
                    if !secondary_index.is_empty() {
                        let index_fields = secondary_index
                            .iter()
                            .filter(|&&idx| idx < table.def.fields.len())
                            .map(|&idx| &table.def.fields[idx])
                            .collect::<Vec<_>>();

                        if !index_fields.is_empty() {
                            let field_names = index_fields
                                .iter()
                                .map(|f| f.name.clone())
                                .collect::<Vec<_>>()
                                .join(", ");

                            let index_name = format!(
                                "idx_{}_{}",
                                table.def.name.to_lowercase(),
                                field_names.replace(", ", "_")
                            );

                            let index_type = match table.def.secondary_index_type {
                                IndexType::Hash => "hash",
                                IndexType::SortedArray => "sortedarray",
                                IndexType::BTree => "btree",
                                IndexType::TTree => "ttree",
                                IndexType::Vector => "vector",
                                IndexType::Json => "json",
                            };

                            let create_index_sql = format!(
                                "CREATE INDEX {} ON {} USING {} ({});\n\n",
                                index_name,
                                table.def.name.to_lowercase(),
                                index_type,
                                field_names
                            );

                            // 写入CREATE INDEX语句
                            file.write_all(create_index_sql.as_bytes())
                                .map_err(|_| RemDbError::FileIoError)?;
                        }
                    }
                }
            }
        }

        // 遍历所有时序表
        for ts_table_id in 0..self.time_series_tables.len() {
            if let Some(ts_table) = &self.time_series_tables[ts_table_id] {
                let def = &ts_table.def;
                let base_def = &def.base;

                // 生成CREATE TIMESERIES TABLE语句，表名转换为小写
                let mut create_ts_table_sql = alloc::string::String::new();
                create_ts_table_sql.push_str(&format!(
                    "CREATE TIMESERIES TABLE {} (\n",
                    base_def.name.to_lowercase()
                ));

                // 添加字段定义
                let mut fields_sql = Vec::new();
                for field in &base_def.fields {
                    let field_sql = format!(
                        "    {} {}",
                        field.name,
                        field.data_type.to_sql_type(field.size)
                    );
                    fields_sql.push(field_sql);
                }

                // 连接字段定义
                create_ts_table_sql.push_str(&fields_sql.join(",\n"));

                // 添加WITH子句
                let mut with_clauses = Vec::new();

                // 添加压缩配置
                let compression_alg = match def.config.compression {
                    crate::time_series::CompressionType::None => "none",
                    crate::time_series::CompressionType::Delta => "delta",
                    crate::time_series::CompressionType::RunLength => "runlength",
                    crate::time_series::CompressionType::DeltaRunLength => "delta-runlength",
                    crate::time_series::CompressionType::DeltaDelta => "delta-delta",
                };
                with_clauses.push(format!(
                    "COMPRESSION = (algorithm='{}', enabled=true)",
                    compression_alg
                ));

                // 添加TTL配置
                let ttl_days = def.config.retention_period_secs / (24 * 3600);
                with_clauses.push(format!("TTL = '{} days'", ttl_days));

                if !with_clauses.is_empty() {
                    create_ts_table_sql
                        .push_str(&format!("\n) WITH {}\n\n", with_clauses.join(", ")));
                } else {
                    create_ts_table_sql.push_str("\n)\n\n");
                }

                // 写入CREATE TIMESERIES TABLE语句
                file.write_all(create_ts_table_sql.as_bytes())
                    .map_err(|_| RemDbError::FileIoError)?;
            }
        }

        Ok(())
    }

    /// 导出数据到文件
    #[cfg(feature = "std")]
    pub fn export_data(&self, path: &str) -> Result<()> {
        // 使用标准库的文件操作
        use std::fs::File;
        use std::io::Write;

        // 先收集所有SQL语句，避免在unsafe块内进行文件I/O
        let mut sql_statements = alloc::string::String::new();

        // 遍历所有表
        for table_id in 0..self.tables.len() {
            if let Some(table) = &self.tables[table_id] {
                // 跳过系统表
                if table.def.name.starts_with("__remdb_") {
                    continue;
                }
                // 遍历表中的所有记录
                let table_ref = table.def.clone();

                // 使用iterate方法遍历记录
                unsafe {
                    table
                        .iterate(|_id, record_ptr| {
                            // 生成INSERT语句，表名转换为小写
                            let mut insert_sql = alloc::string::String::new();
                            insert_sql.push_str(&format!(
                                "INSERT INTO {} (",
                                table_ref.name.to_lowercase()
                            ));

                            // 添加字段名
                            let mut field_names = Vec::new();
                            let mut field_values = Vec::new();

                            for field in table_ref.fields.iter() {
                                field_names.push(field.name.clone());

                                // 获取字段值
                                let field_ptr = record_ptr.add(field.offset);
                                let value_str = match field.data_type {
                                    DataType::UInt8 => format!("{}", *field_ptr as u8),
                                    DataType::UInt16 => format!(
                                        "{}",
                                        core::ptr::read_unaligned(field_ptr as *const u16)
                                    ),
                                    DataType::UInt32 => format!(
                                        "{}",
                                        core::ptr::read_unaligned(field_ptr as *const u32)
                                    ),
                                    DataType::UInt64 => format!(
                                        "{}",
                                        core::ptr::read_unaligned(field_ptr as *const u64)
                                    ),
                                    DataType::Int8 => format!(
                                        "{}",
                                        core::ptr::read_unaligned(field_ptr as *const i8)
                                    ),
                                    DataType::Int16 => format!(
                                        "{}",
                                        core::ptr::read_unaligned(field_ptr as *const i16)
                                    ),
                                    DataType::Int32 => format!(
                                        "{}",
                                        core::ptr::read_unaligned(field_ptr as *const i32)
                                    ),
                                    DataType::Int64 => format!(
                                        "{}",
                                        core::ptr::read_unaligned(field_ptr as *const i64)
                                    ),
                                    DataType::Float32 => format!(
                                        "{}",
                                        core::ptr::read_unaligned(field_ptr as *const f32)
                                    ),
                                    DataType::Float64 => format!(
                                        "{}",
                                        core::ptr::read_unaligned(field_ptr as *const f64)
                                    ),
                                    DataType::Bool => format!("{}", *field_ptr != 0),
                                    DataType::Timestamp => format!(
                                        "{}",
                                        core::ptr::read_unaligned(
                                            field_ptr as *const crate::types::db_timestamp
                                        )
                                        .value
                                    ),
                                    DataType::TimestampTZ => format!(
                                        "{}",
                                        core::ptr::read_unaligned(
                                            field_ptr as *const crate::types::db_timestamp
                                        )
                                        .value
                                    ),
                                    DataType::VarChar | DataType::Char | DataType::Text => {
                                        // 读取字符串并去除尾部的0字节
                                        let mut str_value = alloc::string::String::new();
                                        for i in 0..field.size {
                                            let c = *field_ptr.add(i);
                                            if c == 0 {
                                                break;
                                            }
                                            str_value.push(c as char);
                                        }
                                        alloc::format!("'{}'", str_value)
                                    }
                                    DataType::Interval => {
                                        alloc::format!(
                                            "{}",
                                            core::ptr::read_unaligned(
                                                field_ptr as *const crate::types::db_interval
                                            )
                                            .value
                                        )
                                    }
                                    DataType::Vector => format!("<vector>"),
                                    DataType::Json => format!("<json>"),
                                };

                                field_values.push(value_str);
                            }

                            // 连接字段名和值
                            insert_sql.push_str(&field_names.join(", "));
                            insert_sql.push_str(") VALUES (");
                            insert_sql.push_str(&field_values.join(", "));
                            insert_sql.push_str(");\n");

                            // 将SQL语句添加到集合中
                            sql_statements.push_str(&insert_sql);

                            // 继续遍历
                            true
                        })
                        .expect("iterator should not fail");
                }
            }
        }

        // 现在将所有SQL语句写入文件
        let mut file = File::create(path).map_err(|_| RemDbError::FileIoError)?;
        file.write_all(sql_statements.as_bytes())
            .map_err(|_| RemDbError::FileIoError)?;

        Ok(())
    }

    /// 保存增量快照到文件
    pub fn save_incremental_snapshot(&mut self, path: &str) -> Result<()> {
        // 打开文件 - 使用Write模式
        let handle = crate::platform::file_open(path, crate::platform::FileMode::Write)
            .map_err(|_| RemDbError::FileIoError)?;

        // 使用defer确保文件关闭
        let _defer = Defer::new(|| {
            let _ = crate::platform::file_close(handle);
        });

        // 写入魔数
        let magic = Self::SNAPSHOT_MAGIC.to_le_bytes();
        let written = crate::platform::file_write(handle, magic.as_ptr(), magic.len())
            .map_err(|_| RemDbError::FileIoError)?;
        if written != magic.len() {
            return Err(RemDbError::FileIoError);
        }

        // 写入版本号
        let version = Self::SNAPSHOT_VERSION.to_le_bytes();
        let written = crate::platform::file_write(handle, version.as_ptr(), version.len())
            .map_err(|_| RemDbError::FileIoError)?;
        if written != version.len() {
            return Err(RemDbError::FileIoError);
        }

        // 写入快照类型：1=增量快照
        let snapshot_type = 1u8;
        let written = crate::platform::file_write(handle, &snapshot_type as *const u8, 1)
            .map_err(|_| RemDbError::FileIoError)?;
        if written != 1 {
            return Err(RemDbError::FileIoError);
        }

        // 写入基础版本号（当前全局快照版本号）
        let base_version_bytes = self.snapshot_version.to_le_bytes();
        let written = crate::platform::file_write(
            handle,
            base_version_bytes.as_ptr(),
            base_version_bytes.len(),
        )
        .map_err(|_| RemDbError::FileIoError)?;
        if written != base_version_bytes.len() {
            return Err(RemDbError::FileIoError);
        }

        // 写入表数量
        let table_count = self.tables.len() as u32;
        let table_count_bytes = table_count.to_le_bytes();
        let written = crate::platform::file_write(
            handle,
            table_count_bytes.as_ptr(),
            table_count_bytes.len(),
        )
        .map_err(|_| RemDbError::FileIoError)?;
        if written != table_count_bytes.len() {
            return Err(RemDbError::FileIoError);
        }

        // 写入每个表的增量数据
        for table_id in 0..table_count as usize {
            if let Some(table) = &mut self.tables[table_id] {
                // 计算需要保存的记录数（版本号大于表快照版本的记录）
                let mut changed_records = 0;
                let mut record_indices = Vec::new();

                for i in 0..table.def.max_records {
                    let status_ptr = unsafe { table.get_status_ptr(i) };
                    if unsafe { (*status_ptr).status } == crate::types::RecordStatus::Used {
                        let status = unsafe { &*status_ptr };
                        if status.version > table.snapshot_version as u16 {
                            changed_records += 1;
                            record_indices.push(i);
                        }
                    }
                }

                // 写入表ID（4字节）
                let table_id_u32 = table_id as u32;
                let table_id_bytes = table_id_u32.to_le_bytes();
                let written = crate::platform::file_write(
                    handle,
                    table_id_bytes.as_ptr(),
                    table_id_bytes.len(),
                )
                .map_err(|_| RemDbError::FileIoError)?;
                if written != table_id_bytes.len() {
                    return Err(RemDbError::FileIoError);
                }

                // 写入变化的记录数（4字节）
                let changed_count_u32 = changed_records as u32;
                let changed_count_bytes = changed_count_u32.to_le_bytes();
                let written = crate::platform::file_write(
                    handle,
                    changed_count_bytes.as_ptr(),
                    changed_count_bytes.len(),
                )
                .map_err(|_| RemDbError::FileIoError)?;
                if written != changed_count_bytes.len() {
                    return Err(RemDbError::FileIoError);
                }

                // 动态计算记录大小
                let mut record_size = 0;
                for field in &table.def.fields {
                    record_size += field.size;
                }

                // 写入变化的记录
                for i in record_indices {
                    // 写入记录索引（4字节）
                    let index_u32 = i as u32;
                    let index_bytes = index_u32.to_le_bytes();
                    let written = crate::platform::file_write(
                        handle,
                        index_bytes.as_ptr(),
                        index_bytes.len(),
                    )
                    .map_err(|_| RemDbError::FileIoError)?;
                    if written != index_bytes.len() {
                        return Err(RemDbError::FileIoError);
                    }

                    // 写入记录数据
                    let record_ptr = unsafe { table.get_record_ptr(i) };
                    let written = crate::platform::file_write(handle, record_ptr, record_size)
                        .map_err(|_| RemDbError::FileIoError)?;
                    if written != record_size {
                        return Err(RemDbError::FileIoError);
                    }
                }

                // 更新表快照版本号
                table.snapshot_version = self.snapshot_version;
            }
        }

        // 简化实现，跳过CRC32计算和写入
        Ok(())
    }
}

/// 全局数据库实例 - 使用静态可变变量存储
static mut DB_INSTANCE: Option<RemDb> = None;

/// 初始化数据库全局实例
/// 注意：这是一个简化的实现，实际应用中应该根据需要创建数据库实例
pub fn init_global_db(config: &'static config::DbConfig) -> Result<&'static mut RemDb> {
    unsafe {
        // 无论是否已经初始化过，都创建一个新的数据库实例
        let mut db = RemDb::new(config);

        // 从配置创建表
        for (_table_id, table_def) in config.tables.iter().enumerate() {
            // 创建表
            let table_def_arc = alloc::sync::Arc::new(table_def.clone());
            let table = MemoryTable::new(table_def_arc.clone())?;
            db.tables.push(Some(table));

            // 创建主键索引
            let hash_table_size = (table_def.max_records * 2).next_power_of_two();
            let index_memory_size = crate::index::PrimaryIndex::calculate_memory_size(
                table_def,
                hash_table_size,
                table_def.max_records,
            );
            let index_memory = crate::memory::allocator::alloc(index_memory_size)?;
            let hash_table_start = index_memory.as_ptr()
                as *mut core::option::Option<core::ptr::NonNull<crate::index::PrimaryIndexItem>>;
            let items_start = (index_memory.as_ptr() as usize
                + hash_table_size
                    * core::mem::size_of::<
                        core::option::Option<core::ptr::NonNull<crate::index::PrimaryIndexItem>>,
                    >()) as *mut crate::index::PrimaryIndexItem;
            let primary_index = crate::index::PrimaryIndex::new(
                table_def_arc,
                hash_table_start,
                items_start,
                hash_table_size,
                table_def.max_records,
            );
            db.primary_indices.push(Some(primary_index));
            db.secondary_indices.push(None);
        }

        // 初始化数据库（包括系统表）
        db.init()?;

        // 初始化 DatabaseManager，添加默认数据库
        let _ = db.database_manager.create_database("default", "", None);

        // 将新的数据库实例赋值给 DB_INSTANCE
        // 旧实例会被自动丢弃（Option::Some -> Some 替换时旧值被 drop）
        DB_INSTANCE = Some(db);
        Ok(DB_INSTANCE
            .as_mut()
            .ok_or(RemDbError::UnexpectedNone("DB not initialized"))?)
    }
}

/// 获取全局数据库实例
pub fn get_global_db() -> Option<&'static mut RemDb> {
    unsafe { DB_INSTANCE.as_mut() }
}

/// 重置全局数据库实例
/// 用于测试场景，确保测试之间的隔离
pub fn reset_global_db() {
    unsafe {
        // 关闭HA管理器（仅当ha特性启用时）
        #[cfg(feature = "ha")]
        let _ = crate::ha::shutdown();

        DB_INSTANCE = None;
        // 重置事务管理器状态，包括日志管理器
        let tx_manager = crate::transaction::get_tx_manager();
        tx_manager.reset();
        // 清除日志管理器，确保测试之间的完全隔离
        tx_manager.clear_log_manager();
        // 重置模型管理器，确保测试之间的完全隔离
        let _ = crate::model::model_manager::reset_global_model_manager();
        // 重置全局分配器，确保测试之间的完全隔离
        let _ = crate::memory::allocator::reset_global_allocator();
    }
}

// 导出C接口
#[cfg(feature = "c-api")]
mod c_api;

// Panic handler for no_std environments
#[cfg(all(not(feature = "std"), not(test)))]
#[panic_handler]
fn panic_handler(_info: &core::panic::PanicInfo) -> ! {
    loop {
        core::hint::spin_loop();
    }
}