db.go 18 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461
  1. package DB
  2. import (
  3. "git.ycl.pub/yl32940/TsDB2/DB/core"
  4. "git.ycl.pub/yl32940/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("vrtx: open %s: %w", path, err)
  19. }
  20. return &DB{coreDB: cdb}, nil
  21. }
  22. // OpenWithOptions 打开数据库 (带选项)。ForceInspect 抢救模式: 头页校验失败
  23. // 仍只读打开, 写操作返回 core.ErrReadOnly。
  24. func OpenWithOptions(path string, opt core.OpenOptions) (*DB, error) {
  25. cdb, err := core.OpenWithOptions(path, opt)
  26. if err != nil {
  27. return nil, fmt.Errorf("vrtx: open %s: %w", path, err)
  28. }
  29. return &DB{coreDB: cdb}, nil
  30. }
  31. // Close 关闭数据库
  32. func (db *DB) Close() error {
  33. return db.coreDB.Close()
  34. }
  35. // Commit 显式提交 — 通过 WAL 保证原子性
  36. func (db *DB) Commit() error {
  37. return db.coreDB.Commit()
  38. }
  39. // SetAutoCommitDelay 设置防抖自动提交延迟
  40. // delay=0 表示禁用自动提交 (仅手动 Commit)
  41. // 默认值为 1 秒: 每次写操作重置计时器, 停止操作 1 秒后自动 Commit
  42. func (db *DB) SetAutoCommitDelay(d time.Duration) {
  43. db.coreDB.SetAutoCommitDelay(d)
  44. }
  45. // SetDirtyThreshold 设置脏页阈值(页数),超过此值自动触发 Commit
  46. func (db *DB) SetDirtyThreshold(threshold int) {
  47. db.coreDB.SetDirtyThreshold(threshold)
  48. }
  49. // SetCommitPreset 设置提交策略预设
  50. // preset: "batch" | "online" | "safe" | "ultra-safe"
  51. // - batch: 离线批量导入 (500页, 1s超时) - 高吞吐,可接受少量数据丢失风险
  52. // - online: 在线业务默认 (100页, 500ms超时) - 平衡性能和安全性
  53. // - safe: 金融/医疗关键数据 (20页, 100ms超时) - 最高安全性,性能较低
  54. // - ultra-safe: 极端安全场景 (40页, 50ms超时) - 极致安全,性能很低(慢5-8倍)
  55. //
  56. // 注意: SetCommitPreset 之后**不要再调用 SetAutoCommitDelay(0)** —— delay=0 会
  57. // 完全禁用防抖, 让阈值自动 commit 也无效, 退化成"每行 fsync"。如需调整防抖时长,
  58. // 用 SetAutoCommitDelay(time.Millisecond*200) 这种非零值。
  59. func (db *DB) SetCommitPreset(preset string) {
  60. switch preset {
  61. case "batch":
  62. // 离线批量导入:高吞吐
  63. db.SetDirtyThreshold(500) // 8MB
  64. db.SetAutoCommitDelay(time.Second) // 1s
  65. case "online":
  66. // 在线业务默认:平衡
  67. db.SetDirtyThreshold(100) // 1.6MB
  68. db.SetAutoCommitDelay(500 * time.Millisecond) // 500ms
  69. case "safe":
  70. // 关键数据:最高安全
  71. db.SetDirtyThreshold(20) // 320KB
  72. db.SetAutoCommitDelay(100 * time.Millisecond) // 100ms
  73. case "ultra-safe":
  74. // 极端安全:每40页 Commit,断电最多丢失 ~800行
  75. db.SetDirtyThreshold(40) // 640KB
  76. db.SetAutoCommitDelay(50 * time.Millisecond) // 50ms
  77. default:
  78. // 未知预设,使用默认值
  79. db.SetDirtyThreshold(500)
  80. db.SetAutoCommitDelay(time.Second)
  81. }
  82. }
  83. // LockWrites 锁定写入 —— 锁的是"页落盘"(WritePage), 不是写入动作本身。
  84. //
  85. // 锁定期 Insert/Update/Delete 照常改内存中的页并标记脏, 只有真正要写磁盘时才会阻塞在锁上;
  86. // 没有任何"写入被拒绝"的错误码。Commit/Flush 期间引擎内部会调用它, 目的是拿一个不再变化的
  87. // 磁盘视图。适用场景: 批量导入前、维护操作期间、备份时 (期望这几步之间磁盘内容保持不变)。
  88. func (db *DB) LockWrites() {
  89. db.coreDB.LockWrites()
  90. }
  91. // UnlockWrites 解锁写入 (见 LockWrites 的语义说明)
  92. func (db *DB) UnlockWrites() {
  93. db.coreDB.UnlockWrites()
  94. }
  95. // SetBufferPoolCapacity 设置 BufferPool 容量(页数)
  96. func (db *DB) SetBufferPoolCapacity(capacity int) {
  97. db.coreDB.SetBufferPoolCapacity(capacity)
  98. }
  99. // BeginTransaction 开启新事务
  100. // 返回 Transaction 对象,用于 Commit/Rollback
  101. //
  102. // 使用示例:
  103. //
  104. // txn := db.BeginTransaction()
  105. // defer func() {
  106. // if err := recover(); err != nil {
  107. // txn.Rollback() // 异常时回滚
  108. // panic(err)
  109. // }
  110. // }()
  111. //
  112. // tbl.Insert(&Record{...})
  113. // tbl.Update(...)
  114. //
  115. // if err := txn.Commit(); err != nil {
  116. // return err // 提交失败
  117. // }
  118. func (db *DB) BeginTransaction() *core.Transaction {
  119. return db.coreDB.BeginTransaction()
  120. }
  121. // DebugDBState 打印数据库调试信息 (调试用)
  122. func (db *DB) DebugDBState() {
  123. db.coreDB.DebugDBState()
  124. }
  125. // DebugTableState 打印表调试信息 (调试用)
  126. func (db *DB) DebugTableState(tableName string) {
  127. db.coreDB.DebugTableState(tableName)
  128. }
  129. // Comment 返回数据库介绍区内容 (头页 Comment 字段; 建库时自动写入一段创建说明)
  130. func (db *DB) Comment() string {
  131. return db.coreDB.Comment()
  132. }
  133. // SetComment 改写数据库介绍区 (≤512 字节, 超长报错)
  134. func (db *DB) SetComment(desc string) error {
  135. return db.coreDB.SetComment(desc)
  136. }
  137. // checkNameBudget 在建表 / 建索引之前, 把自动命名额外吃掉的字节预先算清楚。
  138. //
  139. // 为什么必须先算:
  140. // - 表名和索引名各有各的定长上限: 表名 TableNameSize(48)、索引名 IndexNameSize(64);
  141. // 而自动命名要在索引名里再加前缀 —— pk_<表名>、idx_<表名>_<列名> —— 索引名会额外多吃几个字节;
  142. // - 不预检的话, 表会先建成功, 失败发生在紧接着的 CreateIndex 上: 库里留下半成品的表,
  143. // 而错误信息只有"索引名过长", 完全看不出根因其实是表名 / 列名太长。
  144. //
  145. // 报错里把「是哪个名字、多少字节、上限多少」一次讲清楚。
  146. func checkNameBudget(tableName string, indexes []IndexDef) error {
  147. if len(tableName) > core.TableNameSize {
  148. return fmt.Errorf("%w: 表名 %q 有 %d 字节, 上限 %d",
  149. core.ErrTableNameTooLong, tableName, len(tableName), core.TableNameSize)
  150. }
  151. // 主键索引名 pk_<表名> 必须塞进 IndexNameSize。
  152. // 当前取值下 64-48-3 = 13 > 0, 所以这条不会被触发 (文件下方的编译期不变量保证了它);
  153. // 留着是为了将来收紧 IndexNameSize 时, 拦截点仍然落在建表之前,
  154. // 而不是回退成"表建好了、紧跟着建 pk 索引才失败"的半成品状态。
  155. if name := pkIndexPrefix + tableName; len(name) > core.IndexNameSize {
  156. return fmt.Errorf("%w: 自动主键索引名 %q 有 %d 字节 (表名 %q 的 %d 字节 + 前缀 %q), 上限 %d",
  157. core.ErrIndexNameTooLong, name, len(name), tableName, len(tableName), pkIndexPrefix, core.IndexNameSize)
  158. }
  159. for _, idx := range indexes {
  160. if name := idx.indexName(tableName); len(name) > core.IndexNameSize {
  161. return fmt.Errorf("%w: 自动二级索引名 %q 有 %d 字节 (上限 %d) —— 用 `vrtx:\"index:<名字>\"` 显式命名, 或缩短表名 / 首列名",
  162. core.ErrIndexNameTooLong, name, len(name), core.IndexNameSize)
  163. }
  164. }
  165. return nil
  166. }
  167. // 编译期不变量: 表名合法 (<= TableNameSize) 时, 自动主键索引名 pk_<表名> 必须仍然合法,
  168. // 即 IndexNameSize >= TableNameSize + len(pkIndexPrefix)。差值为负时 uint 转换直接编译失败,
  169. // 把这层依赖钉在编译期, 而不是靠注释和自觉。
  170. const _ = uint(core.IndexNameSize - core.TableNameSize - len(pkIndexPrefix))
  171. // UseTable 获取或创建泛型表
  172. // 如果表不存在,会自动创建表和索引
  173. // 如果表已存在,会自动同步 schema:struct 多的字段自动添加,少的字段自动删除
  174. //
  175. // T 必须内嵌 DB.BaseModel / DB.PrimaryModel (主键表) 或 DB.TimeSeriesModel (时序表):
  176. // 内嵌基类既是"这是一张本框架的表"的标记, 也是对象头部增删改查的来源, 所以没有内嵌
  177. // 基类的 T 会被直接拒绝 (ErrNotModel), 不会留下一张半成品表。主键列可由每张表用
  178. // `vrtx:"pk"` 覆盖基类的默认键列 (_id / _ts)。
  179. //
  180. // 用法1: 自动推导表名(推荐)
  181. //
  182. // tbl, _ := DB.UseTable[Employee](db) // 表名自动为 "employee"
  183. //
  184. // 用法2: 手动指定表名
  185. //
  186. // tbl, _ := DB.UseTable[Employee](db, "emp_table") // 自定义表名
  187. func UseTable[T any](db *DB, name ...string) (*Table[T], error) {
  188. var zero T
  189. model, err := ParseModel(&zero)
  190. if err != nil {
  191. return nil, fmt.Errorf("vrtx: parse model: %w", err)
  192. }
  193. // 自动推导表名:从类型名转为小写(如 Employee -> employee)
  194. tableName := typeNameToTableName(zero)
  195. if len(name) > 0 && name[0] != "" {
  196. tableName = name[0] // 用户指定的表名优先
  197. }
  198. // 建表前先把「自动命名要吃掉多少字节」算清楚。pk_ / idx_ 前缀都占用
  199. // core.TableNameSize 的额度, 不提前拦就会出现"表建好了、紧跟着建 pk 索引才失败"
  200. // 的半成品状态, 而且报错只会喊"索引名过长", 把根因 (表名太长) 指偏。
  201. // 表已存在的分支也要查: 新加一个自动命名的二级索引会撞同一个坑。
  202. if err := checkNameBudget(tableName, model.Indexes); err != nil {
  203. return nil, err
  204. }
  205. // 尝试查找已有表
  206. coreTbl, ok := db.coreDB.FindTable(tableName)
  207. if !ok {
  208. // 表不存在,创建新表
  209. coreTbl, err = db.coreDB.CreateTable(tableName, model.CoreFields)
  210. if err != nil {
  211. return nil, fmt.Errorf("vrtx: create table %s: %w", tableName, err)
  212. }
  213. // 创建一级索引
  214. idxName := pkIndexPrefix + tableName
  215. _, err = coreTbl.CreateIndex(idxName, model.GetIndexType(), []byte{model.CorePKField})
  216. if err != nil {
  217. return nil, fmt.Errorf("vrtx: create index for %s: %w", tableName, err)
  218. }
  219. // 自动创建标签声明的二级索引 (同名字段已合并为复合索引)
  220. for _, idx := range model.Indexes {
  221. name := idx.indexName(tableName)
  222. if _, err := coreTbl.CreateIndex(name, idx.IndexType, idx.coreFieldIDs(), idx.Unique); err != nil {
  223. return nil, fmt.Errorf("vrtx: create index %s: %w", name, err)
  224. }
  225. }
  226. } else {
  227. // 表已存在,自动同步 schema
  228. if err := syncSchema(coreTbl, model); err != nil {
  229. return nil, err
  230. }
  231. // 检查并补充缺失的二级索引
  232. for _, idx := range model.Indexes {
  233. name := idx.indexName(tableName)
  234. m, found := coreTbl.IndexMeta().FindIndex(name)
  235. if !found {
  236. if _, err := coreTbl.CreateIndex(name, idx.IndexType, idx.coreFieldIDs(), idx.Unique); err != nil {
  237. return nil, fmt.Errorf("vrtx: create index %s: %w", name, err)
  238. }
  239. continue
  240. }
  241. // 已存在: 校验分量列表与模型一致。
  242. // 不一致说明索引是按旧结构建的 (例如后来才把同名字段改成复合索引),
  243. // 静默沿用会一直用一个"少一列"的索引 —— 查询结果对但索引形同虚设,
  244. // 这里直接报错, 让人决定是重建还是改名。
  245. if !sameIndexFields(m, idx.coreFieldIDs()) {
  246. return nil, fmt.Errorf("vrtx: 索引 %s 已存在但分量列表不同 (已有 %d 列, 模型期望 %d 列), 请先删除该索引",
  247. name, m.FieldCnt, len(idx.coreFieldIDs()))
  248. }
  249. // 模型声明了 unique 而库里那个索引不是唯一的 → 约束根本不会生效。
  250. // 静默沿用等于"模型说唯一、写入却允许重复", 是最危险的一类不一致
  251. // (只在数据脏了以后才发现), 直接报错要求重建。
  252. if idx.Unique && !m.Unique() {
  253. return nil, fmt.Errorf("vrtx: 索引 %s 已存在但不是唯一索引 (模型声明了 unique), 请先删除该索引后重建", name)
  254. }
  255. }
  256. }
  257. tbl := &Table[T]{
  258. db: db,
  259. coreTable: coreTbl,
  260. model: model,
  261. name: tableName,
  262. }
  263. tbl.refreshActiveColumns()
  264. return tbl, nil
  265. }
  266. // sameIndexFields 比较已有索引的分量列表与模型期望是否一致 (顺序敏感)
  267. func sameIndexFields(m *core.IndexMeta, want []byte) bool {
  268. n := int(m.FieldCnt)
  269. if n != len(want) {
  270. return false
  271. }
  272. for i := 0; i < n; i++ {
  273. if m.FieldIDs[i] != want[i] {
  274. return false
  275. }
  276. }
  277. return true
  278. }
  279. // isPKColumn 判断列名是否是主键列。
  280. // 主键列不参与 schema 同步: 它在建表时按模型建好, 一旦被 DropField 删掉,
  281. // 行里的主键值就再也读不回来 —— 引擎的 pkID 只存在于行的 8 字节前缀里,
  282. // 结构体侧的 pk 字段没有任何别的途径反查。
  283. // 主键列既可能是内嵌基类带来的 _id / _ts, 也可能是用户 vrtx:"pk" 指定的列。
  284. func isPKColumn(model *ModelInfo, name string) bool {
  285. if name == pkColumnName || name == tsColumnName {
  286. return true
  287. }
  288. return model.PKInStruct && name == model.PKField
  289. }
  290. // syncSchema 将 DB schema 同步为与 struct 一致
  291. // struct 多 → AddField;struct 少 → DropField
  292. //
  293. // 除了增删列, 还会校验【已存在列的列级参数】: decimal 的 scale 与字符串列的宽度码都必须
  294. // 与模型一致。两者都是"行布局参数":
  295. // - scale 不一致 → "行内字节序 == 数值序"不再成立 (见 types.FieldType_Decimal), 排序静默出错;
  296. // - 宽度不一致 → 模型以为的容量与库里的实际容量不同 (闸门只认库里的档位), 声明 maxlen:16
  297. // 的列实际能写满 1024 字节, 或反之合法值被拒。
  298. //
  299. // 对不上时直接报错, 而不是让新写入的行用另一套参数 —— 那会让同一列里两种布局混排。
  300. func syncSchema(coreTbl *core.Table, model *ModelInfo) error {
  301. // 构建 struct 字段名集合 (不含主键)
  302. structCols := make(map[string]FieldDef, len(model.Fields))
  303. for _, fd := range model.Fields {
  304. if isPKColumn(model, fd.ColName) {
  305. continue
  306. }
  307. structCols[fd.ColName] = fd
  308. }
  309. // 获取 DB 当前活跃列 (排除主键列)
  310. allDBCols := coreTbl.GetActiveFieldNames()
  311. dbCols := make([]string, 0, len(allDBCols))
  312. dbColSet := make(map[string]bool, len(allDBCols))
  313. for _, name := range allDBCols {
  314. if isPKColumn(model, name) {
  315. continue
  316. }
  317. dbCols = append(dbCols, name)
  318. dbColSet[name] = true
  319. }
  320. // 1. struct 有但 DB 没有 → 添加列 (主键列除外: 它由建表时确定, AddField 造不出主键)
  321. for _, fd := range model.Fields {
  322. if isPKColumn(model, fd.ColName) {
  323. continue
  324. }
  325. if !dbColSet[fd.ColName] {
  326. if err := coreTbl.AddField(types.FieldInfo{
  327. FieldType: fd.FieldType,
  328. FieldName: []byte(fd.ColName),
  329. Flag: fd.Flag,
  330. }); err != nil {
  331. return fmt.Errorf("vrtx: add field %s: %w", fd.ColName, err)
  332. }
  333. continue
  334. }
  335. // 已存在的列: 校验列级参数 (decimal 的 scale / 字符串列的宽度码) 是否与模型一致。
  336. dbFT, _ := coreTbl.GetActiveFieldType(fd.ColName)
  337. dbFlag, _ := coreTbl.GetActiveFieldFlag(fd.ColName)
  338. dbParam := dbFlag & types.FlagScaleMask
  339. if fd.FieldType == types.FieldType_Decimal || dbFT == types.FieldType_Decimal {
  340. if fd.FieldType == types.FieldType_Decimal && dbFT == types.FieldType_Decimal &&
  341. dbParam != fd.Param() {
  342. return fmt.Errorf(
  343. "vrtx: 列 %s 的 decimal scale 与库中不一致 (库中 %d, 模型 %d); "+
  344. "同一列只能有一个 scale, 请对齐模型或迁移/重建该列",
  345. fd.ColName, dbParam, fd.Param())
  346. }
  347. continue
  348. }
  349. // 字符串列 (含动态列): 行内宽度必须与模型声明一致。
  350. //
  351. // 宽度决定"这一列占行内哪几字节", 库里的档位就是真正的闸门 (见 types.FieldSize):
  352. // 库里是 1024 字节档而模型声明 16 字节档时, 模型以为"最多 16 字节"的列其实能写满
  353. // 1024; 反过来库里的档位更窄, 模型认为合法的值会被引擎拒。两种都不会自己暴露,
  354. // 所以在这里一次性说清库中与模型各自的宽度和宽度码。
  355. if types.IsStringType(fd.FieldType) || types.IsStringType(dbFT) ||
  356. types.IsDynamic(fd.FieldType) || types.IsDynamic(dbFT) {
  357. dbW := types.FieldSize(dbFT, dbParam)
  358. modelW := fd.Width()
  359. if dbW != modelW {
  360. return fmt.Errorf(
  361. "vrtx: 列 %s 的行内宽度与库中不一致 (库中 %s: %d 字节/宽度码 %d, "+
  362. "模型 %s: %d 字节/宽度码 %d); 定宽随 schema 落盘, 对不上说明模型与库"+
  363. "是两次不同的声明 — 请对齐 maxlen/type, 或重建该列",
  364. fd.ColName, types.TypeNameOf(dbFT), dbW, dbParam,
  365. types.TypeNameOf(fd.FieldType), modelW, fd.Param())
  366. }
  367. }
  368. }
  369. // 2. DB 有但 struct 没有 → 删除列 (dbCols 已排除主键列)
  370. for _, name := range dbCols {
  371. if _, ok := structCols[name]; !ok {
  372. // 错误必须上抛: 静默继续会留下"库里有、struct 没有"的列,
  373. // 后续 UseTable 反复尝试删除, 库与模型的不一致也无从排查。
  374. // (主键列等删不掉的场景由 DropField 守卫转为明确错误。)
  375. if err := coreTbl.DropField(name); err != nil {
  376. return fmt.Errorf("同步 schema 删除列 %q: %w", name, err)
  377. }
  378. }
  379. }
  380. return nil
  381. }
  382. // DropTable 删除表
  383. func (db *DB) DropTable(name string) error {
  384. return db.coreDB.DropTable(name)
  385. }
  386. // CoreDB 返回底层 core.DB (用于高级操作如创建二级索引)
  387. func (db *DB) CoreDB() *core.DB {
  388. return db.coreDB
  389. }
  390. // VACUUM 整库重写, 让数据库文件真正缩小。
  391. //
  392. // 与自动 compact (40% 失效字节触发) 的区别:
  393. // - 自动 compact: 页内墓碑回收, 文件大小几乎不变
  394. // - VACUUM: 整库重写, 文件真正缩小, 归还所有空闲页给 OS
  395. //
  396. // 任何阶段失败 → 删除临时文件, 旧库保持不变。
  397. // 调用后 db 句柄会被关闭并重新打开, 调用者继续使用即可。
  398. func (db *DB) VACUUM(opts ...core.VACUUMOptions) (*core.VACUUMResult, error) {
  399. return db.coreDB.VACUUM(opts...)
  400. }
  401. // SetAutoVACUUM 开启自动 VACUUM, 周期触发。
  402. //
  403. // interval: 两次 VACUUM 的最小间隔 (例如 1*time.Hour)
  404. // options: 每次 VACUUM 使用的选项
  405. //
  406. // 关闭: SetAutoVACUUM(0)
  407. func (db *DB) SetAutoVACUUM(interval time.Duration, opts ...core.VACUUMOptions) {
  408. db.coreDB.SetAutoVACUUM(interval, opts...)
  409. }
  410. // LastVACUUM 返回上次自动 VACUUM 的时间
  411. func (db *DB) LastVACUUM() time.Time {
  412. return db.coreDB.LastVACUUM()
  413. }
  414. // typeNameToTableName 从类型名推导表名(Employee -> employee)
  415. func typeNameToTableName(v any) string {
  416. t := reflect.TypeOf(v)
  417. name := t.Name()
  418. if name == "" {
  419. return "table"
  420. }
  421. // 首字母小写:Employee -> employee, OrderItem -> orderItem
  422. return strings.ToLower(name[:1]) + name[1:]
  423. }