| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461 |
- package DB
- import (
- "git.ycl.pub/yl32940/TsDB2/DB/core"
- "git.ycl.pub/yl32940/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("vrtx: open %s: %w", path, err)
- }
- return &DB{coreDB: cdb}, nil
- }
- // OpenWithOptions 打开数据库 (带选项)。ForceInspect 抢救模式: 头页校验失败
- // 仍只读打开, 写操作返回 core.ErrReadOnly。
- func OpenWithOptions(path string, opt core.OpenOptions) (*DB, error) {
- cdb, err := core.OpenWithOptions(path, opt)
- if err != nil {
- return nil, fmt.Errorf("vrtx: 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倍)
- //
- // 注意: SetCommitPreset 之后**不要再调用 SetAutoCommitDelay(0)** —— delay=0 会
- // 完全禁用防抖, 让阈值自动 commit 也无效, 退化成"每行 fsync"。如需调整防抖时长,
- // 用 SetAutoCommitDelay(time.Millisecond*200) 这种非零值。
- 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 锁定写入 —— 锁的是"页落盘"(WritePage), 不是写入动作本身。
- //
- // 锁定期 Insert/Update/Delete 照常改内存中的页并标记脏, 只有真正要写磁盘时才会阻塞在锁上;
- // 没有任何"写入被拒绝"的错误码。Commit/Flush 期间引擎内部会调用它, 目的是拿一个不再变化的
- // 磁盘视图。适用场景: 批量导入前、维护操作期间、备份时 (期望这几步之间磁盘内容保持不变)。
- func (db *DB) LockWrites() {
- db.coreDB.LockWrites()
- }
- // UnlockWrites 解锁写入 (见 LockWrites 的语义说明)
- 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)
- }
- // Comment 返回数据库介绍区内容 (头页 Comment 字段; 建库时自动写入一段创建说明)
- func (db *DB) Comment() string {
- return db.coreDB.Comment()
- }
- // SetComment 改写数据库介绍区 (≤512 字节, 超长报错)
- func (db *DB) SetComment(desc string) error {
- return db.coreDB.SetComment(desc)
- }
- // checkNameBudget 在建表 / 建索引之前, 把自动命名额外吃掉的字节预先算清楚。
- //
- // 为什么必须先算:
- // - 表名和索引名各有各的定长上限: 表名 TableNameSize(48)、索引名 IndexNameSize(64);
- // 而自动命名要在索引名里再加前缀 —— pk_<表名>、idx_<表名>_<列名> —— 索引名会额外多吃几个字节;
- // - 不预检的话, 表会先建成功, 失败发生在紧接着的 CreateIndex 上: 库里留下半成品的表,
- // 而错误信息只有"索引名过长", 完全看不出根因其实是表名 / 列名太长。
- //
- // 报错里把「是哪个名字、多少字节、上限多少」一次讲清楚。
- func checkNameBudget(tableName string, indexes []IndexDef) error {
- if len(tableName) > core.TableNameSize {
- return fmt.Errorf("%w: 表名 %q 有 %d 字节, 上限 %d",
- core.ErrTableNameTooLong, tableName, len(tableName), core.TableNameSize)
- }
- // 主键索引名 pk_<表名> 必须塞进 IndexNameSize。
- // 当前取值下 64-48-3 = 13 > 0, 所以这条不会被触发 (文件下方的编译期不变量保证了它);
- // 留着是为了将来收紧 IndexNameSize 时, 拦截点仍然落在建表之前,
- // 而不是回退成"表建好了、紧跟着建 pk 索引才失败"的半成品状态。
- if name := pkIndexPrefix + tableName; len(name) > core.IndexNameSize {
- return fmt.Errorf("%w: 自动主键索引名 %q 有 %d 字节 (表名 %q 的 %d 字节 + 前缀 %q), 上限 %d",
- core.ErrIndexNameTooLong, name, len(name), tableName, len(tableName), pkIndexPrefix, core.IndexNameSize)
- }
- for _, idx := range indexes {
- if name := idx.indexName(tableName); len(name) > core.IndexNameSize {
- return fmt.Errorf("%w: 自动二级索引名 %q 有 %d 字节 (上限 %d) —— 用 `vrtx:\"index:<名字>\"` 显式命名, 或缩短表名 / 首列名",
- core.ErrIndexNameTooLong, name, len(name), core.IndexNameSize)
- }
- }
- return nil
- }
- // 编译期不变量: 表名合法 (<= TableNameSize) 时, 自动主键索引名 pk_<表名> 必须仍然合法,
- // 即 IndexNameSize >= TableNameSize + len(pkIndexPrefix)。差值为负时 uint 转换直接编译失败,
- // 把这层依赖钉在编译期, 而不是靠注释和自觉。
- const _ = uint(core.IndexNameSize - core.TableNameSize - len(pkIndexPrefix))
- // UseTable 获取或创建泛型表
- // 如果表不存在,会自动创建表和索引
- // 如果表已存在,会自动同步 schema:struct 多的字段自动添加,少的字段自动删除
- //
- // T 必须内嵌 DB.BaseModel / DB.PrimaryModel (主键表) 或 DB.TimeSeriesModel (时序表):
- // 内嵌基类既是"这是一张本框架的表"的标记, 也是对象头部增删改查的来源, 所以没有内嵌
- // 基类的 T 会被直接拒绝 (ErrNotModel), 不会留下一张半成品表。主键列可由每张表用
- // `vrtx:"pk"` 覆盖基类的默认键列 (_id / _ts)。
- //
- // 用法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("vrtx: parse model: %w", err)
- }
- // 自动推导表名:从类型名转为小写(如 Employee -> employee)
- tableName := typeNameToTableName(zero)
- if len(name) > 0 && name[0] != "" {
- tableName = name[0] // 用户指定的表名优先
- }
- // 建表前先把「自动命名要吃掉多少字节」算清楚。pk_ / idx_ 前缀都占用
- // core.TableNameSize 的额度, 不提前拦就会出现"表建好了、紧跟着建 pk 索引才失败"
- // 的半成品状态, 而且报错只会喊"索引名过长", 把根因 (表名太长) 指偏。
- // 表已存在的分支也要查: 新加一个自动命名的二级索引会撞同一个坑。
- if err := checkNameBudget(tableName, model.Indexes); err != nil {
- return nil, err
- }
- // 尝试查找已有表
- coreTbl, ok := db.coreDB.FindTable(tableName)
- if !ok {
- // 表不存在,创建新表
- coreTbl, err = db.coreDB.CreateTable(tableName, model.CoreFields)
- if err != nil {
- return nil, fmt.Errorf("vrtx: create table %s: %w", tableName, err)
- }
- // 创建一级索引
- idxName := pkIndexPrefix + tableName
- _, err = coreTbl.CreateIndex(idxName, model.GetIndexType(), []byte{model.CorePKField})
- if err != nil {
- return nil, fmt.Errorf("vrtx: create index for %s: %w", tableName, err)
- }
- // 自动创建标签声明的二级索引 (同名字段已合并为复合索引)
- for _, idx := range model.Indexes {
- name := idx.indexName(tableName)
- if _, err := coreTbl.CreateIndex(name, idx.IndexType, idx.coreFieldIDs(), idx.Unique); err != nil {
- return nil, fmt.Errorf("vrtx: create index %s: %w", name, err)
- }
- }
- } else {
- // 表已存在,自动同步 schema
- if err := syncSchema(coreTbl, model); err != nil {
- return nil, err
- }
- // 检查并补充缺失的二级索引
- for _, idx := range model.Indexes {
- name := idx.indexName(tableName)
- m, found := coreTbl.IndexMeta().FindIndex(name)
- if !found {
- if _, err := coreTbl.CreateIndex(name, idx.IndexType, idx.coreFieldIDs(), idx.Unique); err != nil {
- return nil, fmt.Errorf("vrtx: create index %s: %w", name, err)
- }
- continue
- }
- // 已存在: 校验分量列表与模型一致。
- // 不一致说明索引是按旧结构建的 (例如后来才把同名字段改成复合索引),
- // 静默沿用会一直用一个"少一列"的索引 —— 查询结果对但索引形同虚设,
- // 这里直接报错, 让人决定是重建还是改名。
- if !sameIndexFields(m, idx.coreFieldIDs()) {
- return nil, fmt.Errorf("vrtx: 索引 %s 已存在但分量列表不同 (已有 %d 列, 模型期望 %d 列), 请先删除该索引",
- name, m.FieldCnt, len(idx.coreFieldIDs()))
- }
- // 模型声明了 unique 而库里那个索引不是唯一的 → 约束根本不会生效。
- // 静默沿用等于"模型说唯一、写入却允许重复", 是最危险的一类不一致
- // (只在数据脏了以后才发现), 直接报错要求重建。
- if idx.Unique && !m.Unique() {
- return nil, fmt.Errorf("vrtx: 索引 %s 已存在但不是唯一索引 (模型声明了 unique), 请先删除该索引后重建", name)
- }
- }
- }
- tbl := &Table[T]{
- db: db,
- coreTable: coreTbl,
- model: model,
- name: tableName,
- }
- tbl.refreshActiveColumns()
- return tbl, nil
- }
- // sameIndexFields 比较已有索引的分量列表与模型期望是否一致 (顺序敏感)
- func sameIndexFields(m *core.IndexMeta, want []byte) bool {
- n := int(m.FieldCnt)
- if n != len(want) {
- return false
- }
- for i := 0; i < n; i++ {
- if m.FieldIDs[i] != want[i] {
- return false
- }
- }
- return true
- }
- // isPKColumn 判断列名是否是主键列。
- // 主键列不参与 schema 同步: 它在建表时按模型建好, 一旦被 DropField 删掉,
- // 行里的主键值就再也读不回来 —— 引擎的 pkID 只存在于行的 8 字节前缀里,
- // 结构体侧的 pk 字段没有任何别的途径反查。
- // 主键列既可能是内嵌基类带来的 _id / _ts, 也可能是用户 vrtx:"pk" 指定的列。
- func isPKColumn(model *ModelInfo, name string) bool {
- if name == pkColumnName || name == tsColumnName {
- return true
- }
- return model.PKInStruct && name == model.PKField
- }
- // syncSchema 将 DB schema 同步为与 struct 一致
- // struct 多 → AddField;struct 少 → DropField
- //
- // 除了增删列, 还会校验【已存在列的列级参数】: decimal 的 scale 与字符串列的宽度码都必须
- // 与模型一致。两者都是"行布局参数":
- // - scale 不一致 → "行内字节序 == 数值序"不再成立 (见 types.FieldType_Decimal), 排序静默出错;
- // - 宽度不一致 → 模型以为的容量与库里的实际容量不同 (闸门只认库里的档位), 声明 maxlen:16
- // 的列实际能写满 1024 字节, 或反之合法值被拒。
- //
- // 对不上时直接报错, 而不是让新写入的行用另一套参数 —— 那会让同一列里两种布局混排。
- func syncSchema(coreTbl *core.Table, model *ModelInfo) error {
- // 构建 struct 字段名集合 (不含主键)
- structCols := make(map[string]FieldDef, len(model.Fields))
- for _, fd := range model.Fields {
- if isPKColumn(model, fd.ColName) {
- continue
- }
- structCols[fd.ColName] = fd
- }
- // 获取 DB 当前活跃列 (排除主键列)
- allDBCols := coreTbl.GetActiveFieldNames()
- dbCols := make([]string, 0, len(allDBCols))
- dbColSet := make(map[string]bool, len(allDBCols))
- for _, name := range allDBCols {
- if isPKColumn(model, name) {
- continue
- }
- dbCols = append(dbCols, name)
- dbColSet[name] = true
- }
- // 1. struct 有但 DB 没有 → 添加列 (主键列除外: 它由建表时确定, AddField 造不出主键)
- for _, fd := range model.Fields {
- if isPKColumn(model, fd.ColName) {
- continue
- }
- if !dbColSet[fd.ColName] {
- if err := coreTbl.AddField(types.FieldInfo{
- FieldType: fd.FieldType,
- FieldName: []byte(fd.ColName),
- Flag: fd.Flag,
- }); err != nil {
- return fmt.Errorf("vrtx: add field %s: %w", fd.ColName, err)
- }
- continue
- }
- // 已存在的列: 校验列级参数 (decimal 的 scale / 字符串列的宽度码) 是否与模型一致。
- dbFT, _ := coreTbl.GetActiveFieldType(fd.ColName)
- dbFlag, _ := coreTbl.GetActiveFieldFlag(fd.ColName)
- dbParam := dbFlag & types.FlagScaleMask
- if fd.FieldType == types.FieldType_Decimal || dbFT == types.FieldType_Decimal {
- if fd.FieldType == types.FieldType_Decimal && dbFT == types.FieldType_Decimal &&
- dbParam != fd.Param() {
- return fmt.Errorf(
- "vrtx: 列 %s 的 decimal scale 与库中不一致 (库中 %d, 模型 %d); "+
- "同一列只能有一个 scale, 请对齐模型或迁移/重建该列",
- fd.ColName, dbParam, fd.Param())
- }
- continue
- }
- // 字符串列 (含动态列): 行内宽度必须与模型声明一致。
- //
- // 宽度决定"这一列占行内哪几字节", 库里的档位就是真正的闸门 (见 types.FieldSize):
- // 库里是 1024 字节档而模型声明 16 字节档时, 模型以为"最多 16 字节"的列其实能写满
- // 1024; 反过来库里的档位更窄, 模型认为合法的值会被引擎拒。两种都不会自己暴露,
- // 所以在这里一次性说清库中与模型各自的宽度和宽度码。
- if types.IsStringType(fd.FieldType) || types.IsStringType(dbFT) ||
- types.IsDynamic(fd.FieldType) || types.IsDynamic(dbFT) {
- dbW := types.FieldSize(dbFT, dbParam)
- modelW := fd.Width()
- if dbW != modelW {
- return fmt.Errorf(
- "vrtx: 列 %s 的行内宽度与库中不一致 (库中 %s: %d 字节/宽度码 %d, "+
- "模型 %s: %d 字节/宽度码 %d); 定宽随 schema 落盘, 对不上说明模型与库"+
- "是两次不同的声明 — 请对齐 maxlen/type, 或重建该列",
- fd.ColName, types.TypeNameOf(dbFT), dbW, dbParam,
- types.TypeNameOf(fd.FieldType), modelW, fd.Param())
- }
- }
- }
- // 2. DB 有但 struct 没有 → 删除列 (dbCols 已排除主键列)
- for _, name := range dbCols {
- if _, ok := structCols[name]; !ok {
- // 错误必须上抛: 静默继续会留下"库里有、struct 没有"的列,
- // 后续 UseTable 反复尝试删除, 库与模型的不一致也无从排查。
- // (主键列等删不掉的场景由 DropField 守卫转为明确错误。)
- if err := coreTbl.DropField(name); err != nil {
- return fmt.Errorf("同步 schema 删除列 %q: %w", name, err)
- }
- }
- }
- return nil
- }
- // 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
- }
- // VACUUM 整库重写, 让数据库文件真正缩小。
- //
- // 与自动 compact (40% 失效字节触发) 的区别:
- // - 自动 compact: 页内墓碑回收, 文件大小几乎不变
- // - VACUUM: 整库重写, 文件真正缩小, 归还所有空闲页给 OS
- //
- // 任何阶段失败 → 删除临时文件, 旧库保持不变。
- // 调用后 db 句柄会被关闭并重新打开, 调用者继续使用即可。
- func (db *DB) VACUUM(opts ...core.VACUUMOptions) (*core.VACUUMResult, error) {
- return db.coreDB.VACUUM(opts...)
- }
- // SetAutoVACUUM 开启自动 VACUUM, 周期触发。
- //
- // interval: 两次 VACUUM 的最小间隔 (例如 1*time.Hour)
- // options: 每次 VACUUM 使用的选项
- //
- // 关闭: SetAutoVACUUM(0)
- func (db *DB) SetAutoVACUUM(interval time.Duration, opts ...core.VACUUMOptions) {
- db.coreDB.SetAutoVACUUM(interval, opts...)
- }
- // LastVACUUM 返回上次自动 VACUUM 的时间
- func (db *DB) LastVACUUM() time.Time {
- return db.coreDB.LastVACUUM()
- }
- // 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:]
- }
|