package DB import ( "TsDB2/DB/core" "TsDB2/DB/core/types" "encoding/binary" "fmt" "time" ) // Table 泛型表封装。 // // T 必须内嵌 DB.BaseModel / DB.PrimaryModel / DB.TimeSeriesModel —— 由 UseTable 在 // 建表之前强制校验 (见 types.go 的 requireBase), 所以拿到一个 *Table[T] 就意味着 // T 一定有对象头部的增删改查。 type Table[T any] struct { coreTable *core.Table model *ModelInfo name string // 表名 activeColumns map[string]bool // 当前 DB schema 活跃列集合 } // refreshActiveColumns 从 core table 刷新活跃列集合 func (t *Table[T]) refreshActiveColumns() { t.activeColumns = make(map[string]bool) for _, name := range t.coreTable.GetActiveFieldNames() { t.activeColumns[name] = true } } // SchemaColumns 返回当前 DB 中的所有活跃列名 func (t *Table[T]) SchemaColumns() []string { names := make([]string, 0, len(t.activeColumns)) for name := range t.activeColumns { names = append(names, name) } return names } // Insert 插入一条记录 // // 顺带把对象绑定到本表, 这样即使对象是外部构造的, 插完之后也能直接用头部的 // Update / Delete / Reload (见 model.go); 同时把引擎分配的行号带回对象, 之后 // obj.Update() / obj.Delete() 才能精确到"这一行"。 func (t *Table[T]) Insert(obj *T) error { t.bindModel(obj) // 主键显式给了就用它; 没给 (==0) 就向引擎要一个自增号。 // 旧实现无条件用 TotalRows+1 覆盖: 一是调用方填的 _id 被改写 → "按 _id 插入的行 // 按 _id 查不到"; 二是删除会让行数回退 → 号回退并撞上仍存活的 pkID (探针 T4)。 // 现在号一律来自 core 的只增不减游标, 拿到号后再序列化, 保证主键列与 pkID 一致。 // 主键字段可以是内嵌 BaseModel 的 ID, 也可以是用户用 vrtx:"pk" 指定的普通字段。 // // 时序表是唯一的例外, 见下面 IsTimeSeries 分支。 pkID := uint64(0) if t.model.IsTimeSeries { // 时间键不是行号, 两者必须分开: 同一个时间戳下允许有多行 (同一时刻的多个 // 测点 / 多次事件), 这是时序数据的常态。若沿用"时间键即行号", 第二行会被 // 当成重复主键拒掉 —— 数据直接丢, 而这正是引入行号要解决的问题。 // 所以时序表永远向引擎要一个新行号, 时间键保持用户给的原值, 只当索引键用。 pkID = t.coreTable.AllocPKID() if t.model.PKInStruct { ts, err := getPKValue(obj, t.model) if err != nil { return fmt.Errorf("tsdb: insert: %w", err) } if ts == 0 { // 完全没给时间键: 补一个"写入这一刻"的纳秒级时间戳。 // // 时序键就是时间戳本身, 默认值必须与显式传入的时间戳同量纲 (纳秒), // 否则它会成为一个远小于真实时间戳的怪值, 顺序写入下立刻被判为乱序。 // 同一纳秒内连插多行的 TS 会相同, 这是允许的 —— 同刻多行由行号区分。 setPKValue(obj, t.model, uint64(time.Now().UnixNano())) } } } else if t.model.PKInStruct { var err error if pkID, err = getPKValue(obj, t.model); err != nil { return fmt.Errorf("tsdb: insert: %w", err) } if pkID == 0 { pkID = t.coreTable.AllocPKID() setPKValue(obj, t.model, pkID) } } fvs, err := structToFieldValues(obj, t.model, t.activeColumns) if err != nil { return fmt.Errorf("tsdb: insert: %w", err) } // pkID == 0 (结构体没有主键字段) 时由 core 自行分配 assigned, err := t.coreTable.InsertRowWithID(pkID, fvs) if err != nil { return fmt.Errorf("tsdb: insert: %w", err) } setRowIDOf(obj, assigned) return nil } // BatchInsert 批量插入 func (t *Table[T]) BatchInsert(objs []*T) error { for _, obj := range objs { if err := t.Insert(obj); err != nil { return err } } return nil } // Update 更新一条记录 // // 优先按对象携带的引擎行号定位 (从表里读出来或插入过的对象都有), 这样同一时间戳 // 下有多行时改的确实是"自己那一行"; 行号未知时退回按主键列定位 (旧行为)。 func (t *Table[T]) Update(obj *T) error { t.bindModel(obj) fvs, err := structToFieldValues(obj, t.model, t.activeColumns) if err != nil { return fmt.Errorf("tsdb: update: %w", err) } if id := rowIDOf(obj); id != 0 { return t.coreTable.UpdateRowByPK(id, fvs) } pk, err := getPKValue(obj, t.model) if err != nil { return fmt.Errorf("tsdb: update: %w", err) } if pk == 0 { return fmt.Errorf("tsdb: update: primary key is 0") } return t.coreTable.UpdateRow(t.pkIndexName(), encodePKKey(pk), fvs) } // Delete 按主键删除一条记录 func (t *Table[T]) Delete(id uint64) error { return t.coreTable.DeleteRow(t.pkIndexName(), encodePKKey(id)) } // Find 按主键查找一条记录 func (t *Table[T]) Find(id uint64) (*T, error) { // SearchByIndexWithPK 与 SearchByIndex 取的是同一行 (同一条索引键 → 同一次定位), // 只是额外把行的引擎行号带回来 —— 没有它就没法对同刻多行做精确的改 / 删。 pkID, row, err := t.coreTable.SearchByIndexWithPK(t.pkIndexName(), encodePKKey(id)) if err != nil { return nil, err } obj := new(T) if err := fieldValuesToStruct(row, t.model, obj); err != nil { return nil, err } setRowIDOf(obj, pkID) t.bindModel(obj) return obj, nil } // FindAll 返回所有记录 func (t *Table[T]) FindAll() ([]*T, error) { return t.Query().Find() } // Count 返回总行数 func (t *Table[T]) Count() (int64, error) { return t.coreTable.Query().Count() } // Select 流式遍历所有记录 func (t *Table[T]) Select(callback func(*T) bool) error { return t.coreTable.Query().ScanWithPKID(func(pkID uint64, row []types.FieldValue) bool { obj := new(T) if err := fieldValuesToStruct(row, t.model, obj); err != nil { return true } // 行号随手带回对象: 回调里拿到的对象可以直接 Update / Delete, 精确到自己那一行。 setRowIDOf(obj, pkID) t.bindModel(obj) return callback(obj) }) } // Where 开始条件查询 func (t *Table[T]) Where(field string, op string, value interface{}) *Query[T] { q := &Query[T]{table: t} q.addWhere(field, op, value) return q } // OrderBy 排序 func (t *Table[T]) OrderBy(field string, desc ...bool) *Query[T] { q := &Query[T]{table: t} q.OrderBy(field, desc...) return q } // Limit 限制返回行数 func (t *Table[T]) Limit(n int) *Query[T] { q := &Query[T]{table: t} q.Limit(n) return q } // Offset 跳过行数 func (t *Table[T]) Offset(n int) *Query[T] { q := &Query[T]{table: t} q.Offset(n) return q } // Query 创建一个新的查询构建器 func (t *Table[T]) Query() *Query[T] { return &Query[T]{table: t} } func (t *Table[T]) pkIndexName() string { return pkIndexPrefix + t.name } // ===== 模型绑定: 头部增删改查的入口 ===== // Bind 把对象绑定到本表 —— 绑定之后才能用对象头部的 Insert / Update / Delete / // Save / Load / Reload。Insert / Update / Find / FindAll / Select / 条件查询的返回值 // 都会自动绑定, 所以只有"外部手工构造、且第一件事就是 Update/Delete/Reload"的对象 // 需要显式绑一次 (那种情况下用 New 更省事)。 // // 注意: 绑定信息 (属于哪张表、外层对象是谁) 存在对象内部, 因此 // // u2 := *u // 拷贝一个已绑定的对象 // u2.Update() // 更新的是 u, 不是 u2 // // 会更新原对象。需要一份独立的对象请用 Table.New + Load, 或 Table.Find。 func (t *Table[T]) Bind(obj *T) error { if !t.bindModel(obj) { // UseTable 已经把没有基类的 T 挡在门外, 走到这里只可能是 T 的绑定器被特殊的 // 嵌入方式挡住了 (例如把基类嵌在一个指针结构体里)。给同样的错误, 别让它静默失效。 return fmt.Errorf("tsdb: %w: %T 需要内嵌 DB.BaseModel / DB.PrimaryModel / "+ "DB.TimeSeriesModel 才能使用头部方法", ErrNotModel, obj) } return nil } // New 创建一个已经绑定到本表的新对象, 随后可以直接 obj.Load(pk) 读入。 // // 与"手写 &T{} 再 Bind"的区别: 后者要先拿到 T 的类型; 在泛型代码里 (不知道 T 的具体 // 类型时) 只有 Table 能造出 T, New 就是那个入口。 func (t *Table[T]) New() *T { obj := new(T) t.bindModel(obj) return obj } // bindModel 把"这个对象属于哪张表"写回对象内嵌的表基类, 返回是否绑定成功。 // // 走接口断言而不是反射: 每条被读写的记录都要过一遍, 反射在这里是白付的成本。 // 内嵌的表基类会把 modelBinder 提升到 *T 上, 所以一次类型断言就够了。 func (t *Table[T]) bindModel(obj *T) bool { b, ok := any(obj).(modelBinder) if !ok { return false } b.bindTo(t, obj) return true } // ===== session: Table[T] 的类型擦除视图 ===== // // 模型基类不认识 T, 头部方法只能把请求交回这里; 这一层认识 T, 于是把参数断言回 *T。 // 断言失败只可能是"把 A 表的对象交给了 B 表", 所以返回错误而不是 panic。 // sessionObj 断言出 *T, 并顺手重新绑定 —— 对象可能是刚从一个临时副本拷过来的。 func (t *Table[T]) sessionObj(op string, obj any) (*T, error) { o, ok := obj.(*T) if !ok { return nil, fmt.Errorf("tsdb: %s: 表 %s 收到了 %T, 类型不匹配", op, t.name, obj) } t.bindModel(o) return o, nil } func (t *Table[T]) insert(obj any) error { o, err := t.sessionObj("insert", obj) if err != nil { return err } return t.Insert(o) } func (t *Table[T]) update(obj any) error { o, err := t.sessionObj("update", obj) if err != nil { return err } return t.Update(o) } func (t *Table[T]) pkOf(obj any) (uint64, error) { o, err := t.sessionObj("pk", obj) if err != nil { return 0, err } return getPKValue(o, t.model) } // loadInto 把主键 pk 对应的整行读进 obj 指向的对象。 // // 用 *o = *got 整块赋值而不是逐字段搬: 两边是同一个 T, 字段集合一致, 省掉一次反射。 // 代价是 got 的绑定信息也被拷了过来 (self 还指着 got), 所以紧接着必须重新绑定到 o。 func (t *Table[T]) loadInto(pk uint64, obj any) error { o, err := t.sessionObj("load", obj) if err != nil { return err } got, err := t.Find(pk) if err != nil { return err } *o = *got t.bindModel(o) return nil } // reload 重新读一遍 obj 所在的那一行。 // // 行号已知 → 按行号读 (同刻多行也不会读串); 行号未知 → 退回按可见主键读。 func (t *Table[T]) reload(obj any) error { o, err := t.sessionObj("reload", obj) if err != nil { return err } if id := rowIDOf(o); id != 0 { return t.loadByRowID(id, o) } pk, err := getPKValue(o, t.model) if err != nil { return fmt.Errorf("tsdb: reload: %w", err) } return t.loadInto(pk, o) } // loadByRowID 把行号 pkID 对应的整行读进 obj 指向的对象。 func (t *Table[T]) loadByRowID(pkID uint64, obj any) error { o, err := t.sessionObj("load", obj) if err != nil { return err } got, err := t.findByRowID(pkID) if err != nil { return err } *o = *got t.bindModel(o) return nil } // findByRowID 按引擎行号读出一个新对象 (行号是内部句柄, 所以不对外暴露)。 func (t *Table[T]) findByRowID(pkID uint64) (*T, error) { row, err := t.coreTable.FindByPK(pkID) if err != nil { return nil, err } obj := new(T) if err := fieldValuesToStruct(row, t.model, obj); err != nil { return nil, err } setRowIDOf(obj, pkID) t.bindModel(obj) return obj, nil } func (t *Table[T]) deleteByPK(pk uint64) error { return t.Delete(pk) } // deleteByRowID 按引擎行号删除一行 —— 同刻多行时唯一能精确指认"这一行"的方式。 func (t *Table[T]) deleteByRowID(pkID uint64) error { return t.coreTable.DeleteRowByPK(pkID) } func (t *Table[T]) tableName() string { return t.name } func encodePKKey(pk uint64) []byte { buf := make([]byte, 8) binary.BigEndian.PutUint64(buf, pk) return buf }