package DB import ( "git.ycl.pub/yl32940/TsDB2/DB/core" "git.ycl.pub/yl32940/TsDB2/DB/core/types" "encoding/binary" "fmt" "time" ) // Table 泛型表封装。 // // T 必须内嵌 DB.BaseModel / DB.PrimaryModel / DB.TimeSeriesModel —— 由 UseTable 在 // 建表之前强制校验 (见 types.go 的 requireBase), 所以拿到一个 *Table[T] 就意味着 // T 一定有对象头部的增删改查。 // // db 是反向指针, ORM Insert/Update/Delete 自动开 tx (tx.BlobCommit) 时用来从 ORM 层 // 拿 *core.Transaction。UseTable 在创建时填充, 之后只读。 type Table[T any] struct { db *DB 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() 才能精确到"这一行"。 // // Blob 列的落链由 t.db.coreDB (无事务版 BlobCommitter) 承接: 用户不需要手动开事务, // 写入的内容照样自动落链; 用户若已显式开事务, 链页会经 DB.NewPage 登记进该事务, // Rollback 整链回收。这里【不】自动 begin-tx + 立即 Commit —— tx.Commit 只要缓冲池 // 里有脏页 (单行插入必然有) 就走完整 WAL + fsync (~37ms/行), 逐行插入会比防抖提交 // 慢三个数量级, 且 SetCommitPreset 的批量策略被整体绕过 (a56b71d 的性能回退根因)。 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, t.db.coreDB) 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) backfillBlobRefs(obj, t.model, fvs) return nil } // BatchInsert 批量插入 // // 真批量: 走 core 的 InsertRowsWithIDs —— 索引层按"排序灌叶"写入, 数据页/树页 // 落盘按页摊销, saveTableInfo 与防抖提交通知只发一次。灌数吞吐相对逐行 Insert // 提升一个数量级 (逐行路径每行 4 棵树各下降一次 + 每次页保存整页回写)。 // // 两种情形退回逐行 Insert, 语义与旧实现完全一致: // - 时序表: assertTimeOrder 逐行把关写入顺序; // - 任一对象显式携带主键: 逐行查重路径 (显式主键本身就是小众场景)。 func (t *Table[T]) BatchInsert(objs []*T) error { if len(objs) == 0 { return nil } if t.model.IsTimeSeries { for _, obj := range objs { if err := t.Insert(obj); err != nil { return err } } return nil } // 先扫一遍: 有显式主键就整批退回逐行 (避免"半批半单"两条语义) if t.model.PKInStruct { anyExplicit := false for _, obj := range objs { t.bindModel(obj) if pkID, err := getPKValue(obj, t.model); err == nil && pkID != 0 { anyExplicit = true break } } if anyExplicit { for _, obj := range objs { if err := t.Insert(obj); err != nil { return err } } return nil } } pkIDs := make([]uint64, len(objs)) fvs := make([][]types.FieldValue, len(objs)) // 事务边界: 用户已持有活跃事务 (显式 BeginTransaction) 就整体复用 —— 既不顶掉 // 用户事务、也不在批量中途替用户全局提交 (那样用户随后的 Rollback 会失去原子性); // 没有活跃事务才自开自提。core.BeginTransaction 对"已有活跃事务再开"直接 panic, // 所以必须先查 HasActiveTransaction, 顺序不能反。复用期间 blob 列传 coreDB 版 // committer: 有活跃事务时链页经 NewPage 自动登记进该事务, Rollback 一样整链回收。 ownTx := false var tx *core.Transaction var blobc types.BlobCommitter = t.db.coreDB if !t.db.HasActiveTransaction() { tx = t.db.BeginTransaction() ownTx = true blobc = tx } for i, obj := range objs { t.bindModel(obj) // 号先拿到、填进对象再序列化: 行里存的 _id 与引擎行号才一致 (与单行同约定) pkID := t.coreTable.AllocPKID() if t.model.PKInStruct { setPKValue(obj, t.model, pkID) } pkIDs[i] = pkID fv, err := structToFieldValues(obj, t.model, t.activeColumns, blobc) if err != nil { if ownTx { _ = tx.Rollback() } return fmt.Errorf("tsdb: batch insert: %w", err) } fvs[i] = fv } // 号全部来自 AllocPKID 预留号段: 走 InsertRowsReserved 跳过逐行查重 // (走 WithIDs 会被当显式主键逐行 rowStoreHasPK —— 正是单行注释里警告过的 // O(N²) 退化, 5 万行实测多烧 1.2GB 分配) assigned, err := t.coreTable.InsertRowsReserved(pkIDs, fvs) if err != nil { if ownTx { _ = tx.Rollback() } return fmt.Errorf("tsdb: batch insert: %w", err) } if ownTx { if cerr := tx.Commit(); cerr != nil { return fmt.Errorf("tsdb: batch insert: tx commit: %w", cerr) } } for i, obj := range objs { setRowIDOf(obj, assigned[i]) backfillBlobRefs(obj, t.model, fvs[i]) } return nil } // Update 更新一条记录 // // 优先按对象携带的引擎行号定位 (从表里读出来或插入过的对象都有), 这样同一时间戳 // 下有多行时改的确实是"自己那一行"; 行号未知时退回按主键列定位 (旧行为)。 // // Blob 列的落链与旧链回收都由引擎在 UpdateRow 内完成 (新链走 coreDB 的无事务 // BlobCommitter, 旧链走 deferred free), 与 Insert 同一套边界 —— 不自动开 tx。 func (t *Table[T]) Update(obj *T) error { t.bindModel(obj) fvs, err := structToFieldValues(obj, t.model, t.activeColumns, t.db.coreDB) if err != nil { return fmt.Errorf("tsdb: update: %w", err) } if id := rowIDOf(obj); id != 0 { if uerr := t.coreTable.UpdateRowByPK(id, fvs); uerr != nil { return fmt.Errorf("tsdb: update: %w", uerr) } backfillBlobRefs(obj, t.model, fvs) return nil } 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") } if uerr := t.coreTable.UpdateRow(t.pkIndexName(), encodePKKey(pk), fvs); uerr != nil { return fmt.Errorf("tsdb: update: %w", uerr) } backfillBlobRefs(obj, t.model, fvs) return nil } // Delete 按主键删除一条记录 // // 引擎走 heapReleaseRow + releaseRowBlobRefs 回收堆块与 blob 链, 登记进 deferred // free 队列 (有活跃事务时随 tx, 无事务时由 Commit/Close 收尾) —— 不自动开 tx。 func (t *Table[T]) Delete(id uint64) error { return t.coreTable.DeleteRow(t.pkIndexName(), encodePKKey(id)) } // convertRow 反序列化一行 core 值 → *T, 并回写行号 + 绑定表。 // Find / Scan / Select 三个入口共用。 func (t *Table[T]) convertRow(pkID uint64, row []types.FieldValue) (*T, error) { obj := new(T) if err := fieldValuesToStruct(row, t.model, obj, t.coreTable); err != nil { return nil, err } setRowIDOf(obj, pkID) t.bindModel(obj) return obj, nil } // 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, t.coreTable); 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 { var convErr error err := t.coreTable.Query().ScanWithPKID(func(pkID uint64, row []types.FieldValue) bool { obj, err := t.convertRow(pkID, row) if err != nil { // 反序列化失败: 停止扫描并让 Select 返回错误, 不能静默跳过该行 if convErr == nil { convErr = err } return false } return callback(obj) }) if err != nil { return err } return convErr } // 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} } // BlobReader 返回本表的 blob 读取入口 (core.Table, 实现了 types.BlobReader)。 // // 用途: 流式载体 types.Blob 的读回 —— row.Body.Open(tbl.BlobReader()) 拿 // io.ReadCloser, 按流消费 (io.Copy / 逐块), 不把内容整块拉进内存; 适合大对象。 // []byte 载体不需要它 (ORM 读回时自动解链)。 func (t *Table[T]) BlobReader() types.BlobReader { return t.coreTable } 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, t.coreTable); 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 按引擎行号删除一行 —— 同刻多行时唯一能精确指认"这一行"的方式。 // // 自动开 tx 与 Delete 同源: heapReleaseRow / releaseRowBlobRefs 都在 deferred. // // 事务边界同 BatchInsert: 用户已持有活跃事务则复用、绝不替用户提交 (否则 // 用户的 Rollback 会失去原子性); 无活跃事务才自开自提。必须先查 // HasActiveTransaction 再决定 —— core 对"已有活跃事务再开"直接 panic。 func (t *Table[T]) deleteByRowID(pkID uint64) error { ownTx := false var tx *core.Transaction if !t.db.HasActiveTransaction() { tx = t.db.BeginTransaction() ownTx = true } if err := t.coreTable.DeleteRowByPK(pkID); err != nil { if ownTx { _ = tx.Rollback() } return err } if ownTx { if err := tx.Commit(); err != nil { return fmt.Errorf("tsdb: delete: tx commit: %w", err) } } return nil } 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 }