package DB import ( "TsDB2/DB/core" "TsDB2/DB/core/types" "fmt" "reflect" "strings" "time" ) // DB ORM 层数据库入口 type DB struct { coreDB *core.DB } // Open 打开或创建数据库 func Open(path string) (*DB, error) { cdb, err := core.Open(path) if err != nil { return nil, fmt.Errorf("tsdb: open %s: %w", path, err) } return &DB{coreDB: cdb}, nil } // Close 关闭数据库 func (db *DB) Close() error { return db.coreDB.Close() } // Commit 显式提交 — 通过 WAL 保证原子性 func (db *DB) Commit() error { return db.coreDB.Commit() } // SetAutoCommitDelay 设置防抖自动提交延迟 // delay=0 表示禁用自动提交 (仅手动 Commit) // 默认值为 1 秒: 每次写操作重置计时器, 停止操作 1 秒后自动 Commit func (db *DB) SetAutoCommitDelay(d time.Duration) { db.coreDB.SetAutoCommitDelay(d) } // SetDirtyThreshold 设置脏页阈值(页数),超过此值自动触发 Commit func (db *DB) SetDirtyThreshold(threshold int) { db.coreDB.SetDirtyThreshold(threshold) } // SetCommitPreset 设置提交策略预设 // preset: "batch" | "online" | "safe" | "ultra-safe" // - batch: 离线批量导入 (500页, 1s超时) - 高吞吐,可接受少量数据丢失风险 // - online: 在线业务默认 (100页, 500ms超时) - 平衡性能和安全性 // - safe: 金融/医疗关键数据 (20页, 100ms超时) - 最高安全性,性能较低 // - ultra-safe: 极端安全场景 (40页, 50ms超时) - 极致安全,性能很低(慢5-8倍) func (db *DB) SetCommitPreset(preset string) { switch preset { case "batch": // 离线批量导入:高吞吐 db.SetDirtyThreshold(500) // 8MB db.SetAutoCommitDelay(time.Second) // 1s case "online": // 在线业务默认:平衡 db.SetDirtyThreshold(100) // 1.6MB db.SetAutoCommitDelay(500 * time.Millisecond) // 500ms case "safe": // 关键数据:最高安全 db.SetDirtyThreshold(20) // 320KB db.SetAutoCommitDelay(100 * time.Millisecond) // 100ms case "ultra-safe": // 极端安全:每40页 Commit,断电最多丢失 ~800行 db.SetDirtyThreshold(40) // 640KB db.SetAutoCommitDelay(50 * time.Millisecond) // 50ms default: // 未知预设,使用默认值 db.SetDirtyThreshold(500) db.SetAutoCommitDelay(time.Second) } } // LockWrites 锁定写入(禁止 Insert/Update/Delete) // 适用场景:批量导入前、维护操作期间、备份时 func (db *DB) LockWrites() { db.coreDB.LockWrites() } // UnlockWrites 解锁写入(允许 Insert/Update/Delete) func (db *DB) UnlockWrites() { db.coreDB.UnlockWrites() } // SetBufferPoolCapacity 设置 BufferPool 容量(页数) func (db *DB) SetBufferPoolCapacity(capacity int) { db.coreDB.SetBufferPoolCapacity(capacity) } // BeginTransaction 开启新事务 // 返回 Transaction 对象,用于 Commit/Rollback // // 使用示例: // // txn := db.BeginTransaction() // defer func() { // if err := recover(); err != nil { // txn.Rollback() // 异常时回滚 // panic(err) // } // }() // // tbl.Insert(&Record{...}) // tbl.Update(...) // // if err := txn.Commit(); err != nil { // return err // 提交失败 // } func (db *DB) BeginTransaction() *core.Transaction { return db.coreDB.BeginTransaction() } // DebugDBState 打印数据库调试信息 (调试用) func (db *DB) DebugDBState() { db.coreDB.DebugDBState() } // DebugTableState 打印表调试信息 (调试用) func (db *DB) DebugTableState(tableName string) { db.coreDB.DebugTableState(tableName) } // UseTable 获取或创建泛型表 // 如果表不存在,会自动创建表和索引 // 如果表已存在,会自动同步 schema:struct 多的字段自动添加,少的字段自动删除 // // 用法1: 自动推导表名(推荐) // // tbl, _ := DB.UseTable[Employee](db) // 表名自动为 "employee" // // 用法2: 手动指定表名 // // tbl, _ := DB.UseTable[Employee](db, "emp_table") // 自定义表名 func UseTable[T any](db *DB, name ...string) (*Table[T], error) { var zero T model, err := ParseModel(&zero) if err != nil { return nil, fmt.Errorf("tsdb: parse model: %w", err) } // 自动推导表名:从类型名转为小写(如 Employee -> employee) tableName := typeNameToTableName(zero) if len(name) > 0 && name[0] != "" { tableName = name[0] // 用户指定的表名优先 } // 尝试查找已有表 coreTbl, ok := db.coreDB.FindTable(tableName) if !ok { // 表不存在,创建新表 coreTbl, err = db.coreDB.CreateTable(tableName, model.CoreFields) if err != nil { return nil, fmt.Errorf("tsdb: create table %s: %w", tableName, err) } // 创建一级索引 idxName := "pk_" + tableName _, err = coreTbl.CreateIndex(idxName, model.GetIndexType(), []byte{model.CorePKField}) if err != nil { return nil, fmt.Errorf("tsdb: create index for %s: %w", tableName, err) } } else { // 表已存在,自动同步 schema syncSchema(coreTbl, model) } tbl := &Table[T]{ coreTable: coreTbl, model: model, name: tableName, } tbl.refreshActiveColumns() return tbl, nil } // syncSchema 将 DB schema 同步为与 struct 一致 // struct 多 → AddField;struct 少 → DropField func syncSchema(coreTbl *core.Table, model *ModelInfo) { // 构建 struct 字段名集合 (不含自动主键) structCols := make(map[string]FieldDef, len(model.Fields)) for _, fd := range model.Fields { structCols[fd.ColName] = fd } // 获取 DB 当前活跃列 (排除自动主键 _id / _ts) dbCols := coreTbl.GetActiveFieldNames() dbColSet := make(map[string]bool, len(dbCols)) for _, name := range dbCols { if name == "_id" || name == "_ts" { continue } dbColSet[name] = true } // 1. struct 有但 DB 没有 → 添加列 for _, fd := range model.Fields { if !dbColSet[fd.ColName] { coreTbl.AddField(types.FieldInfo{ FieldType: fd.FieldType, FieldName: []byte(fd.ColName), }) } } // 2. DB 有但 struct 没有 → 删除列 for _, name := range dbCols { if name == "_id" || name == "_ts" { continue } if _, ok := structCols[name]; !ok { coreTbl.DropField(name) } } } // DropTable 删除表 func (db *DB) DropTable(name string) error { return db.coreDB.DropTable(name) } // CoreDB 返回底层 core.DB (用于高级操作如创建二级索引) func (db *DB) CoreDB() *core.DB { return db.coreDB } // typeNameToTableName 从类型名推导表名(Employee -> employee) func typeNameToTableName(v any) string { t := reflect.TypeOf(v) name := t.Name() if name == "" { return "table" } // 首字母小写:Employee -> employee, OrderItem -> orderItem return strings.ToLower(name[:1]) + name[1:] }