db.go 5.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211
  1. package DB
  2. import (
  3. "TsDB2/DB/core"
  4. "TsDB2/DB/core/types"
  5. "fmt"
  6. "reflect"
  7. "strings"
  8. "time"
  9. )
  10. // DB ORM 层数据库入口
  11. type DB struct {
  12. coreDB *core.DB
  13. }
  14. // Open 打开或创建数据库
  15. func Open(path string) (*DB, error) {
  16. cdb, err := core.Open(path)
  17. if err != nil {
  18. return nil, fmt.Errorf("tsdb: open %s: %w", path, err)
  19. }
  20. return &DB{coreDB: cdb}, nil
  21. }
  22. // Close 关闭数据库
  23. func (db *DB) Close() error {
  24. return db.coreDB.Close()
  25. }
  26. // Commit 显式提交 — 通过 WAL 保证原子性
  27. func (db *DB) Commit() error {
  28. return db.coreDB.Commit()
  29. }
  30. // SetAutoCommitDelay 设置防抖自动提交延迟
  31. // delay=0 表示禁用自动提交 (仅手动 Commit)
  32. // 默认值为 1 秒: 每次写操作重置计时器, 停止操作 1 秒后自动 Commit
  33. func (db *DB) SetAutoCommitDelay(d time.Duration) {
  34. db.coreDB.SetAutoCommitDelay(d)
  35. }
  36. // SetDirtyThreshold 设置脏页阈值(页数),超过此值自动触发 Commit
  37. func (db *DB) SetDirtyThreshold(threshold int) {
  38. db.coreDB.SetDirtyThreshold(threshold)
  39. }
  40. // SetCommitPreset 设置提交策略预设
  41. // preset: "batch" | "online" | "safe"
  42. // - batch: 离线批量导入 (500页, 1s超时) - 高吞吐,可接受少量数据丢失风险
  43. // - online: 在线业务默认 (100页, 500ms超时) - 平衡性能和安全性
  44. // - safe: 金融/医疗关键数据 (20页, 100ms超时) - 最高安全性,性能较低
  45. func (db *DB) SetCommitPreset(preset string) {
  46. switch preset {
  47. case "batch":
  48. // 离线批量导入:高吞吐
  49. db.SetDirtyThreshold(500) // 8MB
  50. db.SetAutoCommitDelay(time.Second) // 1s
  51. case "online":
  52. // 在线业务默认:平衡
  53. db.SetDirtyThreshold(100) // 1.6MB
  54. db.SetAutoCommitDelay(500 * time.Millisecond) // 500ms
  55. case "safe":
  56. // 关键数据:最高安全
  57. db.SetDirtyThreshold(20) // 320KB
  58. db.SetAutoCommitDelay(100 * time.Millisecond) // 100ms
  59. default:
  60. // 未知预设,使用默认值
  61. db.SetDirtyThreshold(500)
  62. db.SetAutoCommitDelay(time.Second)
  63. }
  64. }
  65. // LockWrites 锁定写入(禁止 Insert/Update/Delete)
  66. // 适用场景:批量导入前、维护操作期间、备份时
  67. func (db *DB) LockWrites() {
  68. db.coreDB.LockWrites()
  69. }
  70. // UnlockWrites 解锁写入(允许 Insert/Update/Delete)
  71. func (db *DB) UnlockWrites() {
  72. db.coreDB.UnlockWrites()
  73. }
  74. // SetBufferPoolCapacity 设置 BufferPool 容量(页数)
  75. func (db *DB) SetBufferPoolCapacity(capacity int) {
  76. db.coreDB.SetBufferPoolCapacity(capacity)
  77. }
  78. // DebugDBState 打印数据库调试信息 (调试用)
  79. func (db *DB) DebugDBState() {
  80. db.coreDB.DebugDBState()
  81. }
  82. // DebugTableState 打印表调试信息 (调试用)
  83. func (db *DB) DebugTableState(tableName string) {
  84. db.coreDB.DebugTableState(tableName)
  85. }
  86. // UseTable 获取或创建泛型表
  87. // 如果表不存在,会自动创建表和索引
  88. // 如果表已存在,会自动同步 schema:struct 多的字段自动添加,少的字段自动删除
  89. //
  90. // 用法1: 自动推导表名(推荐)
  91. //
  92. // tbl, _ := DB.UseTable[Employee](db) // 表名自动为 "employee"
  93. //
  94. // 用法2: 手动指定表名
  95. //
  96. // tbl, _ := DB.UseTable[Employee](db, "emp_table") // 自定义表名
  97. func UseTable[T any](db *DB, name ...string) (*Table[T], error) {
  98. var zero T
  99. model, err := ParseModel(&zero)
  100. if err != nil {
  101. return nil, fmt.Errorf("tsdb: parse model: %w", err)
  102. }
  103. // 自动推导表名:从类型名转为小写(如 Employee -> employee)
  104. tableName := typeNameToTableName(zero)
  105. if len(name) > 0 && name[0] != "" {
  106. tableName = name[0] // 用户指定的表名优先
  107. }
  108. // 尝试查找已有表
  109. coreTbl, ok := db.coreDB.FindTable(tableName)
  110. if !ok {
  111. // 表不存在,创建新表
  112. coreTbl, err = db.coreDB.CreateTable(tableName, model.CoreFields)
  113. if err != nil {
  114. return nil, fmt.Errorf("tsdb: create table %s: %w", tableName, err)
  115. }
  116. // 创建一级索引
  117. idxName := "pk_" + tableName
  118. _, err = coreTbl.CreateIndex(idxName, model.GetIndexType(), []byte{model.CorePKField})
  119. if err != nil {
  120. return nil, fmt.Errorf("tsdb: create index for %s: %w", tableName, err)
  121. }
  122. } else {
  123. // 表已存在,自动同步 schema
  124. syncSchema(coreTbl, model)
  125. }
  126. tbl := &Table[T]{
  127. coreTable: coreTbl,
  128. model: model,
  129. name: tableName,
  130. }
  131. tbl.refreshActiveColumns()
  132. return tbl, nil
  133. }
  134. // syncSchema 将 DB schema 同步为与 struct 一致
  135. // struct 多 → AddField;struct 少 → DropField
  136. func syncSchema(coreTbl *core.Table, model *ModelInfo) {
  137. // 构建 struct 字段名集合 (不含自动主键)
  138. structCols := make(map[string]FieldDef, len(model.Fields))
  139. for _, fd := range model.Fields {
  140. structCols[fd.ColName] = fd
  141. }
  142. // 获取 DB 当前活跃列 (排除自动主键 _id / _ts)
  143. dbCols := coreTbl.GetActiveFieldNames()
  144. dbColSet := make(map[string]bool, len(dbCols))
  145. for _, name := range dbCols {
  146. if name == "_id" || name == "_ts" {
  147. continue
  148. }
  149. dbColSet[name] = true
  150. }
  151. // 1. struct 有但 DB 没有 → 添加列
  152. for _, fd := range model.Fields {
  153. if !dbColSet[fd.ColName] {
  154. coreTbl.AddField(types.FieldInfo{
  155. FieldType: fd.FieldType,
  156. FieldName: []byte(fd.ColName),
  157. })
  158. }
  159. }
  160. // 2. DB 有但 struct 没有 → 删除列
  161. for _, name := range dbCols {
  162. if name == "_id" || name == "_ts" {
  163. continue
  164. }
  165. if _, ok := structCols[name]; !ok {
  166. coreTbl.DropField(name)
  167. }
  168. }
  169. }
  170. // DropTable 删除表
  171. func (db *DB) DropTable(name string) error {
  172. return db.coreDB.DropTable(name)
  173. }
  174. // CoreDB 返回底层 core.DB (用于高级操作如创建二级索引)
  175. func (db *DB) CoreDB() *core.DB {
  176. return db.coreDB
  177. }
  178. // typeNameToTableName 从类型名推导表名(Employee -> employee)
  179. func typeNameToTableName(v any) string {
  180. t := reflect.TypeOf(v)
  181. name := t.Name()
  182. if name == "" {
  183. return "table"
  184. }
  185. // 首字母小写:Employee -> employee, OrderItem -> orderItem
  186. return strings.ToLower(name[:1]) + name[1:]
  187. }