db.go 33 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906
  1. package core
  2. import (
  3. "git.ycl.pub/yl32940/TsDB2/DB/core/index"
  4. "git.ycl.pub/yl32940/TsDB2/DB/core/page"
  5. "git.ycl.pub/yl32940/TsDB2/DB/core/types"
  6. "git.ycl.pub/yl32940/TsDB2/DB/core/util"
  7. "encoding/binary"
  8. "errors"
  9. "fmt"
  10. "strings"
  11. "sync"
  12. "sync/atomic"
  13. "time"
  14. "unsafe"
  15. )
  16. type TableMeta struct {
  17. Name string
  18. FieldCount uint32
  19. TotalRows uint64
  20. FirstHeap uint32
  21. LastHeap uint32
  22. FirstIndex uint32
  23. LastIndex uint32
  24. FirstData uint32
  25. LastData uint32
  26. RowSize uint16
  27. }
  28. type DB struct {
  29. header *HeaderPage
  30. tables *TablePage
  31. file *File
  32. // path 数据库文件路径 (VACUUM / 自动重打开使用)。
  33. // 由 Open(path) 赋值, 运行时不变。
  34. path string
  35. // 防抖自动提交
  36. autoCommitDelay time.Duration
  37. autoCommitTimer *time.Timer
  38. autoCommitMu sync.Mutex
  39. // commitMu 提交互斥: 防抖定时器 / 脏页阈值回调 / 用户显式提交 / 事务提交共用。
  40. // 只串行化"提交"这一步, 不参与写入路径 —— 详见 DB.Commit 的说明。
  41. commitMu sync.Mutex
  42. // accMu 操作级可重入串行锁: 引擎不支持并发读写, 所有公开操作入口都过它,
  43. // 把多 goroutine 的误并发自动串行化(防止改坏结构 / 读到中间态)。
  44. // 显式事务在 Begin 持有、Commit/Rollback 释放; 事务内同 goroutine 重入不阻塞。
  45. // 详见 access_lock.go。
  46. accMu accessLock
  47. // 当前活跃事务(nil 表示无事务)。
  48. // 用 atomic.Pointer: 写线程 (Begin/Commit/Rollback) 与自动提交的定时器
  49. // goroutine (autoCommit/NotifyWrite) 会并发读写它, 裸指针是 data race。
  50. // 注意这只保证【指针本身】的可见性 —— Transaction 内部字段仍受
  51. // "单写入者模型"约束 (同一时刻只允许一个 goroutine 跑写入路径)。
  52. currentTxn atomic.Pointer[Transaction]
  53. // pendingMu 保护 pendingFree / pendingFreeChain: 写入路径登记、提交路径出队,
  54. // 两条路径并发 (写线程插入/删除 + 后台提交线程) 时必须串行化, 否则切片并发读写
  55. // 会丢登记项 / 触发数据竞争。
  56. pendingMu sync.Mutex
  57. // pendingFree 无事务上下文时积压的"待回收页"(堆页摘链 / 批量删除后作废的旧 DataPage)。
  58. // 不在修改中途直写盘, 而是等 Commit 成功 / Close 后再真正还链 —— 见 DeferFreePage 的说明。
  59. // currentTxn != nil 时改入 txn.pendingFree (Rollback 时整体丢弃)。
  60. pendingFree []uint32
  61. // pendingFreeChain 无事务上下文时积压的"待回收整链"(见 DeferFreeChain)。
  62. // currentTxn != nil 时改入 txn.pendingFreeChain (Rollback 时整体丢弃)。
  63. pendingFreeChain []pendingChain
  64. // freePage 真正把页还给空闲链表 (直写盘), 供上面的队列出队时调用
  65. freePage func(pageID uint32)
  66. // freeChain 真正把整链还给空闲链表 (直写盘), 供上面的队列出队时调用
  67. freeChain func(head, tail, count uint32) error
  68. // 自动 VACUUM 调度 (SetAutoVACUUM 启用)
  69. autoVacuumMu sync.Mutex
  70. autoVacuumTimer *time.Timer
  71. autoVacuumEnabled bool
  72. autoVacuumOpts VACUUMOptions
  73. lastVACUUM time.Time
  74. // vacuumMu 串行化 VACUUM 本体: 手动与自动调度可能并发触发,
  75. // 双开会对同一文件做 Close/Open 交错 (自动调度抢到锁才跑, 否则本轮跳过)。
  76. vacuumMu sync.Mutex
  77. // dirtyThreshold 脏页阈值的用户配置 (内存态, wireRuntime 重绑阈值回调时用)。
  78. dirtyThreshold int
  79. }
  80. // pendingChain 一条待回收的整链: head…tail 共 count 页, 链内 next 在 PageHdr.NextPage
  81. type pendingChain struct {
  82. head, tail, count uint32
  83. }
  84. // OpenOptions 打开数据库的选项。
  85. type OpenOptions struct {
  86. // ForceInspect 抢救模式: 头页校验 (Magic/XXH32) 失败时仍打开 —— 打印警告、
  87. // 只读 (一切写操作返回 ErrReadOnly)、跳过 WAL 恢复 (恢复会写盘)。
  88. // 用途: 头页损坏后抢救数据 / analyzer 类工具手工勘验。
  89. ForceInspect bool
  90. // ReadOnly 只读打开: 文件以 O_RDONLY 打开、加【共享】跨进程锁 (可与其它只读
  91. // 进程共存, 与读写进程互斥)、跳过 WAL 恢复与一切写盘。用于报表 / 分析等不
  92. // 修改数据的连接 (DSN mode=ro)。
  93. ReadOnly bool
  94. }
  95. // Open 打开数据库 (等价 OpenWithOptions 零值)。
  96. func Open(path string) (*DB, error) {
  97. return OpenWithOptions(path, OpenOptions{})
  98. }
  99. // OpenWithOptions 打开数据库。新建库走 newHeader + PrepareWrite (Version=11,
  100. // CRC 随首次落盘生成); 已有库先 Validate (Magic 必查, v11 且 CRC==0 的旧库
  101. // 跳过, 其余校验 XXH32), 校验失败默认拒绝打开, ForceInspect 时只读抢救。
  102. func OpenWithOptions(path string, opt OpenOptions) (*DB, error) {
  103. var (
  104. ff *File
  105. err error
  106. )
  107. if opt.ReadOnly {
  108. ff, err = OpenFileReadOnly(path)
  109. } else {
  110. ff, err = OpenFile(path)
  111. }
  112. if err != nil {
  113. return nil, err
  114. }
  115. // 启用页面缓冲池 (只读时仅作读缓存, 不会产生脏页)
  116. ff.InitBufferPool(DefaultBufferPoolCapacity)
  117. stat, err := ff.f.Stat()
  118. if err != nil {
  119. return nil, err
  120. }
  121. if opt.ReadOnly && stat.Size() == 0 {
  122. _ = ff.f.Close()
  123. return nil, fmt.Errorf("只读打开失败: 数据库文件 %q 不存在或为空", path)
  124. }
  125. // 自适应缓冲池容量: 按文件总页数 1/4 配置 —— 整库可放内存时最大化命中率,
  126. // 巨型库也不会无限制占内存。上界 8000 页 (128MB) 截断, 下界保留默认值。
  127. // 旧固定 16MB 对 214MB+ 的真实库欠配, 1M 行 Delete 全程冷读导致 ~50% 时间
  128. // 花在 pread 上; 扩到 1/4 后, 全表 COUNT(*) / 随机点查的主数据页基本全驻。
  129. if stat.Size() > 0 {
  130. filePages := int(stat.Size() / int64(PageSize))
  131. auto := filePages / 4
  132. if auto > 8000 {
  133. auto = 8000
  134. }
  135. if auto > DefaultBufferPoolCapacity {
  136. ff.SetBufferPoolCapacity(auto)
  137. }
  138. }
  139. // 跨进程文件锁: 只读用共享锁 (多个只读可共存), 读写用排他锁; 在读头页 /
  140. // WAL 恢复之前先占锁, 阻止读写进程与读者/写者同时打开。
  141. if stat.Size() > 0 {
  142. var lockErr error
  143. if opt.ReadOnly {
  144. lockErr = ff.tryRLock()
  145. } else {
  146. lockErr = ff.tryLock()
  147. }
  148. if lockErr != nil {
  149. _ = ff.f.Close() // 释放句柄 (锁随之释放)
  150. return nil, lockErr
  151. }
  152. }
  153. var header *HeaderPage
  154. if stat.Size() == 0 {
  155. // 建库: 未提供介绍 → 头页 Comment 默认写入 VersionComment
  156. // (当前版本的功能介绍与修复点); 建库时间见 CreateTime 字段
  157. header, err = newHeader("")
  158. if err != nil {
  159. return nil, err
  160. }
  161. if err := ff.WriteHeader(header); err != nil {
  162. return nil, err
  163. }
  164. // 新库头页写入后文件已有内容, 立即加跨进程文件锁。
  165. if err := ff.tryLock(); err != nil {
  166. _ = ff.f.Close()
  167. return nil, err
  168. }
  169. } else {
  170. header, err = ff.ReadHeader()
  171. if err != nil {
  172. return nil, err
  173. }
  174. // 头页校验: Magic 必查; CRC 对 v11-旧库 (HeaderCrc==0) 跳过, 其余校验。
  175. if err := header.Validate(); err != nil {
  176. // 只读打开本就不写; 抢救模式显式要求 —— 两者都降级为只读, 其余情况拒绝。
  177. if !opt.ForceInspect && !opt.ReadOnly {
  178. _ = ff.f.Close() // 校验失败必须释放文件句柄, 否则 Windows 上文件被占用
  179. return nil, err
  180. }
  181. fmt.Printf("警告: %v\n以只读模式打开 (写操作被禁用)\n", err)
  182. ff.readOnly = true
  183. }
  184. // 版本闸门: 低于当前版本的库结构对不上, 一律拒绝 (抢救模式也不例外)
  185. if header.Version < Version {
  186. _ = ff.f.Close()
  187. return nil, fmt.Errorf("不兼容的数据库文件版本 %d (当前支持 >= %d), 请重建数据库", header.Version, Version)
  188. }
  189. }
  190. // ==========================
  191. // 初始化表页
  192. // ==========================
  193. // realFreePage 真正把页还给空闲链表。
  194. // 必须直接写盘: 走 BufferPool 会被 ClearCache 清掉; 且该页在缓存里通常仍是
  195. // "脏"的旧内容, 不覆盖的话 Commit 收集脏页时会把刚写好的空闲链表指针反向
  196. // 覆盖掉, 导致空闲链错乱/页被重复分配。
  197. realFreePage := func(pageID uint32) {
  198. ff.withHeader(func() {
  199. header.recyclePage(pageID, ff.WritePageDirect)
  200. syncBuf := make([]byte, PageSize)
  201. _ = ff.readPageDisk(pageID, syncBuf)
  202. ff.SyncCache(pageID, syncBuf)
  203. _ = ff.writeHeaderLocked(header)
  204. })
  205. }
  206. // realFreeChain 把一条已用 PageHdr.NextPage 串好的整链一次性还链 (O(1))。
  207. //
  208. // 与 realFreePage 的区别只是"链 vs 单页": 这里只改写一条链尾页 + 更新头页,
  209. // 代价与链长无关。逐页遍历回收 N 页需要 2N 次 IO (每页一次读 + 一次写),
  210. // 拼接只要 3 次 —— DropTable 的堆链、WAL 提交后的 WAL 链都走这里。
  211. realFreeChain := func(head, tail, count uint32) error {
  212. var err error
  213. ff.withHeader(func() {
  214. err = recycleChainWithCache(ff, header, head, tail, count)
  215. if err != nil {
  216. return
  217. }
  218. err = ff.writeHeaderLocked(header)
  219. })
  220. return err
  221. }
  222. // 启动期的句柄用 nil 接收者构造: Init 阶段只会用到 ReadPage/SavePage,
  223. // 涉及 db 的闭包有 nil 守卫且此阶段不会触发; db 构造完成后以真实接收者重建
  224. // (wireRuntime), 闭包从此绑定本句柄 —— 见 buildHandleDb 的说明。
  225. tablePage := &TablePage{
  226. handle: (*DB)(nil).buildHandleDb(ff, header, realFreePage, realFreeChain),
  227. }
  228. // ==========================
  229. // 初始化索引元数据页
  230. // ==========================
  231. var db *DB
  232. saveHeader := false
  233. //如果索引页和表页都是空的话则直接创建
  234. if header.TablePage == 0 {
  235. if opt.ReadOnly {
  236. _ = ff.f.Close()
  237. return nil, fmt.Errorf("只读打开的数据库缺少表页 (TablePage=0): 不是有效数据库, 无法在只读模式初始化")
  238. }
  239. // 用带读盘能力的分配: 即便此时空闲链非空 (例如 TablePage 丢失的旧库),
  240. // 也能正确从空闲链摘页; 新库空闲链为空, 走 NextFreePage 递增分配。
  241. header.TablePage = ff.allocPage(header, true)
  242. tablePage.InitHeader(header.TablePage)
  243. saveHeader = true
  244. }
  245. if saveHeader {
  246. err = ff.WriteHeader(header)
  247. if err != nil {
  248. return nil, err
  249. }
  250. }
  251. // ==========================
  252. // WAL 恢复: 检查是否有未完成的 WAL 事务
  253. // ==========================
  254. // WAL 恢复会写盘 —— 抢救模式 (只读) 下跳过: 待恢复事务保持挂起, 数据页
  255. // 仍是盘上原状, 读路径不受影响 (这正是抢救模式要的效果)。
  256. //
  257. // 必须在 tablePage.Init 之前: 恢复会把已提交事务的表信息页镜像 (TotalRows /
  258. // 堆链指针 / RowSize 等) 重放回磁盘, 而 Init 会把这些元数据【载入内存 Table
  259. // 对象】。先 Init 再恢复, 内存里就是旧元数据: 崩溃恢复后 TotalRows 对不上、
  260. // 堆链指针过期, 后续写入按过期结构分配 = 损坏 (wal_crash_chaos 压测实锤)。
  261. if !ff.readOnly {
  262. if err := RecoverWAL(ff, header); err != nil {
  263. _ = ff.f.Close() // 恢复失败必须释放文件句柄, 否则 Windows 上文件被占用
  264. return nil, err
  265. }
  266. }
  267. // 恢复完成后再载入表元数据: 上面重放过的表页此刻才是最新状态
  268. tablePage.Init(header.TablePage)
  269. db = &DB{
  270. file: ff,
  271. header: header,
  272. tables: tablePage,
  273. path: path,
  274. autoCommitDelay: DefaultAutoCommitDelay,
  275. freePage: realFreePage,
  276. freeChain: realFreeChain,
  277. dirtyThreshold: DefaultDirtyThreshold,
  278. }
  279. // 反向注入, TablePage.Create 时拿到 db 指针; bindDB 顺带给 Init 阶段已经从盘上
  280. // 加载出来的表补上 db (那时 DB 还没构造完, 只能现在回填)。
  281. db.tables.bindDB(db)
  282. db.startAutoCommit()
  283. // 运行期回调与句柄闭包以真实接收者重绑 (启动期那版绑定的是 nil 接收者)。
  284. db.wireRuntime()
  285. return db, nil
  286. }
  287. // buildHandleDb 构造 TablePage 的运行期句柄, 闭包绑定接收者 db。
  288. //
  289. // Open 在 db 存在之前先用 nil 接收者构造一版 (Init 阶段用不到 db 闭包,
  290. // 且涉及 db 的闭包都有 nil 守卫); db 构造完成后以真实接收者重建 —— 闭包
  291. // 从此绑定本句柄的 currentTxn/pendingFree。VACUUM 的 adoptEngineState 换过
  292. // 引擎状态后也必须用它重绑, 否则闭包仍指向旧句柄: 新句柄上的事务不会被
  293. // markTouched (Rollback 静默失效), 新分配的页登记到旧句柄 (回滚后成孤儿页)。
  294. func (db *DB) buildHandleDb(ff *File, header *HeaderPage,
  295. realFreePage func(uint32), realFreeChain func(uint32, uint32, uint32) error) page.HandleDb {
  296. return page.HandleDb{
  297. NewPage: func() uint32 {
  298. // allocPage(erase=true): headerMu 内摘空闲链 / 递增游标并推进头页; 擦页在
  299. // 锁外走 WritePage —— 调用方紧接着的整页写入会覆盖池里同一份缓存, 因此不
  300. // 额外产生物理 IO。读空闲链直接读盘 (绕过 BufferPool)。
  301. pid := ff.allocPage(header, true)
  302. // 事务内新分配的页登记到事务: 分配已就地推进 header 游标 (头页延迟到
  303. // Commit 落盘, 不可回退), Rollback 时必须把这些页还给空闲链, 否则成为
  304. // 永久泄漏的孤儿页。
  305. if db != nil {
  306. if txn := db.currentTxn.Load(); txn != nil {
  307. txn.allocated = append(txn.allocated, pid)
  308. }
  309. }
  310. return pid
  311. },
  312. FreePage: func(pageID uint32) {
  313. // 事务内被立即还链的页, 从 allocated 撤销登记, 避免 Rollback 时二次归还。
  314. if db != nil {
  315. if txn := db.currentTxn.Load(); txn != nil {
  316. txn.undoAllocated(pageID)
  317. }
  318. }
  319. realFreePage(pageID)
  320. },
  321. // 延迟回收: 堆页摘链 / 旧 DataPage 作废时只登记, 等提交成功后再还链。
  322. // 中途直写盘会让"页已清零"先于 WAL 落盘 —— 崩溃或 Rollback 后原页变空页,
  323. // 而堆链/索引仍指向它。事务内入 txn.pendingFree (回滚即丢弃),
  324. // 事务外入 db.pendingFree (下次提交/关闭时统一还链)。
  325. DeferFreePage: func(pageID uint32) {
  326. if db == nil {
  327. realFreePage(pageID) // 启动期(如 WAL 恢复前)没有事务上下文, 保持原行为
  328. return
  329. }
  330. if txn := db.currentTxn.Load(); txn != nil {
  331. txn.pendingFree = append(txn.pendingFree, pageID)
  332. return
  333. }
  334. db.pendingMu.Lock()
  335. db.pendingFree = append(db.pendingFree, pageID)
  336. db.pendingMu.Unlock()
  337. },
  338. ReadPage: ff.ReadPage,
  339. SavePage: ff.WritePage,
  340. // 整链回收: DropTable 的堆链 / WAL 提交后的 WAL 链 / 哈希索引的桶目录链与桶
  341. // 溢出链, 都是「首尾和链长已知、且链内 next 与空闲链同用 PageHdr.NextPage」
  342. // 的链 —— 一次拼接 O(1) 还链, 不必逐页遍历 (逐页 = 每页读+写各一次)。
  343. FreeChain: realFreeChain,
  344. // 延迟整链回收: 与 DeferFreePage 同一时序约定 (提交成功后才真正还链),
  345. // 但一条链只登记一次, 提交后一次 FreeChain 还链。
  346. DeferFreeChain: func(head, tail, count uint32) {
  347. if head == 0 || count == 0 {
  348. return
  349. }
  350. if db == nil {
  351. _ = realFreeChain(head, tail, count) // 启动期没有事务上下文, 保持原行为
  352. return
  353. }
  354. c := pendingChain{head: head, tail: tail, count: count}
  355. if txn := db.currentTxn.Load(); txn != nil {
  356. txn.pendingFreeChain = append(txn.pendingFreeChain, c)
  357. return
  358. }
  359. db.pendingMu.Lock()
  360. db.pendingFreeChain = append(db.pendingFreeChain, c)
  361. db.pendingMu.Unlock()
  362. },
  363. // ValidPage: 页号合法性 = 落在已分配区间 [2, NextFreePage)。
  364. // 0 是头页、1 是建库预留的表页; >= NextFreePage 的页从未分配过, 文件里没有
  365. // 它的位置 —— 放进空闲链, 下一次 NewPage 就会拿着它写到几十 GB 偏移处
  366. // (实测 DROP TABLE 假死 52s、稀疏文件暴涨到 40GB 的直接原因)。
  367. ValidPage: func(pageID uint32) bool {
  368. ff.headerMu.Lock()
  369. limit := header.NextFreePage
  370. ff.headerMu.Unlock()
  371. return pageID >= 2 && pageID < limit
  372. },
  373. }
  374. }
  375. // errIfReadOnly 抢救模式 (force-inspect) 下写入入口的统一闸门。
  376. // File 层的 readOnly 守卫保证"任何数据都不可能落盘"; 这里的入口闸门让写入
  377. // 在第一时间就拿到清晰的 ErrReadOnly, 而不是缓存一串脏页到提交时才失败。
  378. func (db *DB) errIfReadOnly() error {
  379. if db.file.readOnly {
  380. return ErrReadOnly
  381. }
  382. return nil
  383. }
  384. // errIfReadOnly 表级写入入口的闸门 (经 table.db 转发)。
  385. func (table *Table) errIfReadOnly() error {
  386. if table.db == nil {
  387. return nil
  388. }
  389. return table.db.errIfReadOnly()
  390. }
  391. // wireRuntime 把运行期回调绑定到本句柄: 事务页登记 (onPageWrite)、脏页阈值
  392. // 自动提交回调、表页句柄闭包。VACUUM 的 adoptEngineState 换引擎状态后必须重绑,
  393. // 否则这些闭包仍指向旧句柄。
  394. func (db *DB) wireRuntime() {
  395. db.file.SetOnPageWriteCallback(func(pageID uint32) {
  396. if txn := db.currentTxn.Load(); txn != nil {
  397. txn.markTouched(pageID)
  398. }
  399. })
  400. // 脏页阈值: 超过时自动触发 Commit 刷盘, 防止内存暴涨。
  401. // 走 autoCommit: 有活跃事务时不能从中途提交 (会破坏事务原子性)。
  402. db.file.SetDirtyThreshold(db.dirtyThreshold, db.autoCommit)
  403. db.tables.handle = db.buildHandleDb(db.file, db.header, db.freePage, db.freeChain)
  404. }
  405. // NewPage 分配一个新页号, 走空闲链 / 头页游标的同一路径, 并按事务内已分配登记.
  406. //
  407. // 这条路径与 HandleDb.NewPage 闭包完全等价 —— 后者只绑给 TablePage 的几个调用点,
  408. // 其它地方 (如 BlobCommit) 不持有 handle, 必须走这里. 返回的是页号, 调用方拿到后
  409. // 仍要自己用 db.file.ReadPage/WritePage 读写内容.
  410. func (db *DB) NewPage() uint32 {
  411. pid := db.file.allocPage(db.header, true)
  412. if txn := db.currentTxn.Load(); txn != nil {
  413. txn.allocated = append(txn.allocated, pid)
  414. }
  415. return pid
  416. }
  417. func (db *DB) updateTime() {
  418. db.file.withHeader(func() {
  419. db.header.ModifyTime = time.Now().UnixNano()
  420. _ = db.file.writeHeaderLocked(db.header)
  421. })
  422. }
  423. // Comment 返回头页里的数据库介绍区内容 (建库时自动写入一段创建说明, 之后可用
  424. // SetComment 改写)。返回的是去掉尾部零字节后的字符串。
  425. func (db *DB) Comment() string {
  426. db.file.headerMu.Lock()
  427. comment := db.header.Comment
  428. db.file.headerMu.Unlock()
  429. return strings.TrimRight(string(comment[:]), "\x00")
  430. }
  431. // SetComment 改写数据库介绍区 (头页 Comment 字段, ≤512 字节, 超长报错)。
  432. // 写入走 withHeader + writeHeaderLocked (与 updateTime 同模式), 并更新 ModifyTime。
  433. func (db *DB) SetComment(desc string) error {
  434. if err := db.errIfReadOnly(); err != nil {
  435. return err
  436. }
  437. defer db.lockOp()()
  438. if len(desc) > len(db.header.Comment) {
  439. return fmt.Errorf("描述字符串长度 %d 超过最大限制 %d", len(desc), len(db.header.Comment))
  440. }
  441. var werr error
  442. db.file.withHeader(func() {
  443. var comment [PageSize4K - 332]byte
  444. copy(comment[:], desc)
  445. db.header.Comment = comment
  446. db.header.ModifyTime = time.Now().UnixNano()
  447. werr = db.file.writeHeaderLocked(db.header)
  448. })
  449. return werr
  450. }
  451. func (db *DB) FindTable(name string) (*Table, bool) {
  452. tbl, ok := db.tables.Find(name)
  453. if ok && tbl.onWrite == nil {
  454. tbl.SetOnWrite(db.NotifyWrite)
  455. }
  456. return tbl, ok
  457. }
  458. func (db *DB) CreateTable(name string, fields []types.FieldInfo) (*Table, error) {
  459. if db.HasActiveTransaction() {
  460. return nil, errDDLInTransaction("CreateTable")
  461. }
  462. defer db.lockOp()() // DDL 全程串行; 内部 db.Commit 同 goroutine 重入
  463. tbl, err := db.tables.Create(name, fields)
  464. if err != nil {
  465. return nil, err
  466. }
  467. tbl.SetOnWrite(db.NotifyWrite)
  468. db.file.withHeader(func() {
  469. db.header.TableCount++
  470. _ = db.file.writeHeaderLocked(db.header)
  471. })
  472. db.updateTime()
  473. // DDL 隐式提交: 把新表钉到盘上 —— DDL 不具有事务性 (与 Oracle/MySQL 一致),
  474. // 未提交的 DDL 随回滚消失后, 内存元数据 (TableInfo/IndexMeta) 会指向已回
  475. // 收的页, 后续读写立即迷路 (txn_crash_chaos 压测实锤)
  476. if err := db.Commit(); err != nil {
  477. return nil, err
  478. }
  479. return tbl, nil
  480. }
  481. // errDDLInTransaction 事务内执行 DDL 的明确拒绝。
  482. //
  483. // DDL 一旦允许进入事务, 回滚它就会出现"内存元数据 vs 磁盘真值"的分叉:
  484. // Rollback 按页丢弃缓存, 而 Table.info / IndexMeta.metas 是独立于页缓存的
  485. // Go 堆副本, 不会随缓存一起消失 —— 未提交的建表/建索引回滚后, 它们仍指向
  486. // 事务中分配、已被回收的页号, 下一个操作立刻 load failed / 写穿到空闲页。
  487. // 因此这里宁可显式拒绝, 也不让调用方踩进这个不可恢复的坑。
  488. func errDDLInTransaction(op string) error {
  489. return fmt.Errorf("%s: DDL 操作不支持在事务中执行 (DDL 为隐式提交语义, 请先 Commit 或 Rollback 当前事务)", op)
  490. }
  491. // ddlBoundary DDL 收尾: 无活跃事务时隐式提交, 有活跃事务时应已被入口拒绝
  492. // (防御性再查一次)。返回 error 时 DDL 已成功但落盘失败 —— 调用方应视 DDL 为未完成。
  493. func (db *DB) ddlBoundary(op string) error {
  494. if db.HasActiveTransaction() {
  495. return errDDLInTransaction(op)
  496. }
  497. return db.Commit()
  498. }
  499. func (db *DB) DropTable(name string) error {
  500. if db.HasActiveTransaction() {
  501. return errDDLInTransaction("DropTable")
  502. }
  503. defer db.lockOp()() // DDL 全程串行; 内部 Flush/Commit 同 goroutine 重入
  504. // 先刷脏页再清缓存: ClearCache 会把缓存里的页统统标成"干净",
  505. // 而随后的 FlushBufferPool(CollectDirty) 会把它们直接移出缓存 ——
  506. // 如果缓存里还有尚未落盘的页(手动提交模式下很常见), 就会被静默丢掉。
  507. _ = db.file.FlushBufferPool()
  508. db.file.ClearCache()
  509. if err := db.tables.Drop(name); err != nil {
  510. return err
  511. }
  512. db.file.withHeader(func() {
  513. db.header.TableCount--
  514. _ = db.file.writeHeaderLocked(db.header)
  515. })
  516. db.updateTime()
  517. // 先刷盘再清缓存, 防止 Drop 中标记的 Flag(0x80) 丢失
  518. _ = db.file.FlushBufferPool()
  519. db.file.ClearCache()
  520. return db.ddlBoundary("DropTable")
  521. }
  522. // TruncateTable 清空一张表的全部数据但保留表结构与索引定义 (DDL, 隐式提交)。
  523. // 事务内拒绝 (与其它 DDL 一致)。
  524. func (db *DB) TruncateTable(name string) error {
  525. if err := db.errIfReadOnly(); err != nil {
  526. return err
  527. }
  528. if db.HasActiveTransaction() {
  529. return errDDLInTransaction("TruncateTable")
  530. }
  531. defer db.lockOp()()
  532. if err := db.tables.Truncate(name); err != nil {
  533. return err
  534. }
  535. return db.ddlBoundary("TruncateTable")
  536. }
  537. func (db *DB) GetAllTables() []TableMeta {
  538. res := make([]TableMeta, 0, len(db.tables.tables))
  539. for _, t := range db.tables.tables {
  540. if util.IsHighBitSet(t.info.Flag) {
  541. continue
  542. }
  543. res = append(res, TableMeta{
  544. Name: util.BytesToString(t.info.Name[:]),
  545. FieldCount: uint32(t.info.FieldCount),
  546. TotalRows: t.info.TotalRows,
  547. FirstHeap: t.info.FirstHeap,
  548. LastHeap: t.info.LastHeap,
  549. FirstData: t.info.FirstData,
  550. LastData: t.info.LastData,
  551. RowSize: t.info.RowSize,
  552. })
  553. }
  554. return res
  555. }
  556. func (db *DB) Close() error {
  557. defer db.lockOp()() // 等待在飞操作结束, 并在关闭全程挡住新操作
  558. // 停止防抖定时器
  559. db.stopAutoCommit()
  560. // 只读实例: 无脏页/无待写, 直接关句柄 —— Commit/写头/fsync 在只读 fd 上
  561. // (Windows) 会报 Access is denied, 且没有任何需要持久化的内容。
  562. if db.file.readOnly {
  563. return db.file.Close()
  564. }
  565. // 先提交剩余脏页 (走 WAL, 提交路径会把头页一并落盘)
  566. errCommit := db.Commit()
  567. // 头页兜底: Commit 失败时游标变更还悬着, 必须先于缓冲池刷盘落下,
  568. // 否则盘上出现页号超过 NextFreePage 的脏页, 重开后会被二次分配覆盖
  569. errHeader := db.file.FlushHeader(db.header)
  570. // BufferPool 安全网 (理论上已无脏页); 写盘失败必须带出去 —— 失败页仍脏,
  571. // 但 Close 之后无人再刷, 吞掉错误等于静默丢数据
  572. errFlush := db.file.FlushBufferPool()
  573. // 积压的待回收页: 此时链指针已落盘, 可以安全还链
  574. db.drainPendingFree()
  575. // 收尾 fsync: 保证 Header / 数据页真正落到持久介质,
  576. // 否则进程退出后立刻断电可能丢掉最后一次提交
  577. errSync := db.file.sync()
  578. errClose := db.file.Close()
  579. // 任何一个环节出错都要带出去 (errors.Join 忽略 nil)
  580. return errors.Join(errCommit, errHeader, errFlush, errSync, errClose)
  581. }
  582. // freeNow 把一批待回收页真正还给空闲链表 (直写盘)
  583. func (db *DB) freeNow(ids []uint32) {
  584. for _, pid := range ids {
  585. if db.freePage != nil {
  586. db.freePage(pid)
  587. }
  588. }
  589. }
  590. // freeChainNow 把一批待回收整链真正还给空闲链表 (每条链一次 O(1) 拼接)
  591. func (db *DB) freeChainNow(chains []pendingChain) {
  592. for _, c := range chains {
  593. if db.freeChain != nil {
  594. _ = db.freeChain(c.head, c.tail, c.count)
  595. }
  596. }
  597. }
  598. // drainPendingFree 回收无事务上下文时积压的页与整链。
  599. // 只能在"链指针已经持久化"的时刻调用 —— 即提交成功之后, 或 Close 收尾时。
  600. func (db *DB) drainPendingFree() {
  601. // 先原子地摘下当前队列 (写入路径可能正在并发登记, 只取走此刻已入队的部分),
  602. // 再在锁外真正还链 —— 还链会走 headerMu + 磁盘 IO, 不该占着 pendingMu。
  603. db.pendingMu.Lock()
  604. ids := db.pendingFree
  605. db.pendingFree = nil
  606. chains := db.pendingFreeChain
  607. db.pendingFreeChain = nil
  608. db.pendingMu.Unlock()
  609. if len(ids) == 0 && len(chains) == 0 {
  610. return
  611. }
  612. db.freeNow(ids)
  613. db.freeChainNow(chains)
  614. }
  615. // Commit 显式提交 — 通过 WAL 保证原子性
  616. // 同时重置防抖定时器
  617. //
  618. // commitMu 保证同一时刻只有一笔提交在飞: 防抖定时器、脏页阈值回调(在写入 goroutine
  619. // 内同步触发) 和用户显式调用都可能撞在一起, 而两笔提交同时收集/回写同一批脏页会把
  620. // 页的归属搞乱 (前一笔的快照会把后一笔刚改出来的内容判成"已落盘")。
  621. // 这把锁只串行化提交本身, 不参与写入路径: 写入只在 BufferPool.WritePage 里等
  622. // writeLock (见 LockWrites), 两者没有嵌套获取, 因此不会互相死锁。
  623. func (db *DB) Commit() error {
  624. if err := db.errIfReadOnly(); err != nil {
  625. return err
  626. }
  627. defer db.lockOp()() // 操作级串行: 提交期间挡住其他 goroutine 的读写(独立调用时)
  628. db.commitMu.Lock()
  629. defer db.commitMu.Unlock()
  630. // 取消待执行的自动提交
  631. db.autoCommitMu.Lock()
  632. if db.autoCommitTimer != nil {
  633. db.autoCommitTimer.Stop()
  634. }
  635. db.autoCommitMu.Unlock()
  636. // 1. 收集所有 BufferPool 脏页 (TablePage + IndexPage + DataPage 等)
  637. dirty := db.file.CollectDirty()
  638. if len(dirty) == 0 {
  639. // 无脏页 = 修改已持久, 积压的待回收页可以安全还链
  640. db.drainPendingFree()
  641. return nil // 无脏页, 跳过
  642. }
  643. // 2. 创建事务
  644. wal := db.BeginTxn()
  645. // 3. 记录所有脏页
  646. for _, d := range dirty {
  647. wal.RecordPage(d.PageID, d.Data)
  648. }
  649. // 4. HeaderPage 也纳入 (pageID=0)
  650. // marshalHeader 要读整张头页 (含提交路径正在改的 WAL 链指针), 必须在 headerMu 内拍快照。
  651. db.file.withHeader(func() {
  652. wal.RecordPage(0, marshalHeader(db.header))
  653. })
  654. // 5. 原子提交
  655. if err := wal.Commit(); err != nil {
  656. // 提交失败: 这批页必须继续留在缓存里当脏页。CollectDirty 现在不再把它们移出
  657. // 缓存, 所以正常情况下它们本来就在; RestoreDirty 只是兜底 (错误路径上页回收
  658. // 等动作也可能碰过这些页), 而且只补缺失条目, 不会用旧快照盖掉更新的内容。
  659. db.file.RestoreDirty(dirty)
  660. return err
  661. }
  662. // 6. 提交成功 → 把这一批页收成干净页。
  663. // 必须按快照的版本号逐页判定: 提交窗口里写入线程可能已经把某些页又改了一遍
  664. // (缓存里是更新的内容, 本次 WAL 记的却是快照那份), 那些页要继续留脏, 否则新
  665. // 内容既不会被写盘, 脏标记还报告"已持久化" —— 该页一旦被淘汰就永久丢失。
  666. // (旧实现这里是 ClearAll, 无差别地把整个缓存标干净, 恰好会抹掉这种页。)
  667. db.file.MarkClean(dirty)
  668. // 7. 提交成功 → 链指针已持久化, 积压的待回收页现在可以真正还链
  669. db.drainPendingFree()
  670. return nil
  671. }
  672. // ======================== 防抖自动提交 ========================
  673. // SetAutoCommitDelay 设置防抖自动提交延迟
  674. // delay=0 表示禁用自动提交 (仅手动 Commit)
  675. func (db *DB) SetAutoCommitDelay(d time.Duration) {
  676. db.autoCommitMu.Lock()
  677. defer db.autoCommitMu.Unlock()
  678. db.autoCommitDelay = d
  679. if db.autoCommitTimer != nil {
  680. db.autoCommitTimer.Stop()
  681. db.autoCommitTimer = nil
  682. }
  683. }
  684. // NotifyWrite 写操作通知 (InsertRow/DeleteRow/UpdateRow 完成后调用)
  685. // 重置防抖定时器: 如果延迟期间没有新操作, 则自动 Commit
  686. func (db *DB) NotifyWrite() {
  687. if db.autoCommitDelay <= 0 {
  688. return
  689. }
  690. // 有活跃事务时不启动自动提交: 否则定时器到点会把事务从中途截断提交,
  691. // 破坏事务原子性, 并使后续 Rollback 无法真正撤销
  692. if db.currentTxn.Load() != nil {
  693. return
  694. }
  695. db.autoCommitMu.Lock()
  696. defer db.autoCommitMu.Unlock()
  697. // 复用同一个 timer 对象,避免频繁创建/销毁
  698. if db.autoCommitTimer == nil {
  699. // 首次创建
  700. db.autoCommitTimer = time.AfterFunc(db.autoCommitDelay, db.autoCommit)
  701. } else {
  702. // 后续重置(无需重新分配对象,减少 GC 压力)
  703. db.autoCommitTimer.Reset(db.autoCommitDelay)
  704. }
  705. }
  706. // autoCommit 自动提交入口 (防抖定时器 / 脏页阈值共用)
  707. // 有活跃事务时直接跳过, 避免把事务从中途截断提交
  708. func (db *DB) autoCommit() {
  709. if db.currentTxn.Load() != nil {
  710. return
  711. }
  712. _ = db.Commit()
  713. }
  714. // startAutoCommit 启动防抖自动提交
  715. func (db *DB) startAutoCommit() {
  716. // 不立即启动定时器, 等待第一次 NotifyWrite
  717. }
  718. // stopAutoCommit 停止防抖自动提交
  719. func (db *DB) stopAutoCommit() {
  720. db.autoCommitMu.Lock()
  721. defer db.autoCommitMu.Unlock()
  722. if db.autoCommitTimer != nil {
  723. db.autoCommitTimer.Stop()
  724. db.autoCommitTimer = nil
  725. }
  726. }
  727. // SetDirtyThreshold 设置脏页阈值(页数),超过此值自动触发 Commit
  728. func (db *DB) SetDirtyThreshold(threshold int) {
  729. db.dirtyThreshold = threshold
  730. db.file.SetDirtyThreshold(threshold, db.autoCommit)
  731. }
  732. // LockWrites 锁定写入 —— 实际锁的是"页落盘"(BufferPool.WritePage 会 waitForWriteLock)。
  733. // 锁定期 Insert/Update/Delete 仍可改内存中的页并标记脏, 只是要写磁盘时阻塞; 没有
  734. // "写入被拒绝"的错误码。Commit/Flush 期间由引擎内部调用, 以取得稳定的磁盘视图。
  735. func (db *DB) LockWrites() {
  736. db.file.LockWrites()
  737. }
  738. // UnlockWrites 解锁写入 (见 LockWrites 的语义说明)
  739. func (db *DB) UnlockWrites() {
  740. db.file.UnlockWrites()
  741. }
  742. // SetBufferPoolCapacity 设置 BufferPool 容量(页数)
  743. func (db *DB) SetBufferPoolCapacity(capacity int) {
  744. db.file.SetBufferPoolCapacity(capacity)
  745. }
  746. // marshalHeader 将 HeaderPage 序列化为字节切片 (填充到 PageSize)
  747. func marshalHeader(h *HeaderPage) []byte {
  748. buf := make([]byte, PageSize)
  749. // 在【副本】上重算 CRC 再快照: 内存中的 HeaderCrc 可能滞后于游标 / 链
  750. // 指针的内存内变更 (allocPage 只标脏不落盘), 快照字节必须与快照内容自洽,
  751. // 否则崩溃恢复重放出头页后 CRC 与内容对不上。
  752. snap := *h
  753. snap.PrepareWrite()
  754. src := (*[PageSize4K]byte)(unsafe.Pointer(&snap))[:]
  755. copy(buf, src)
  756. return buf
  757. }
  758. // ======================== 调试诊断 ========================
  759. // DebugDBState 打印数据库调试信息
  760. func (db *DB) DebugDBState() {
  761. db.file.withHeader(func() {
  762. fmt.Printf("[DEBUG] Header: TablePage=%d NextFreePage=%d FreePageStart=%d FirstWAL=%d LastWAL=%d NextTxn=%d\n",
  763. db.header.TablePage, db.header.NextFreePage, db.header.FreePageStart,
  764. db.header.FirstWALPage, db.header.LastWALPage, db.header.NextTxnID)
  765. })
  766. fmt.Printf("[DEBUG] BufferPool: total=%d dirty=%d\n", db.file.bp.Len(), db.file.bp.DirtyCount())
  767. }
  768. // DebugTableState 打印表调试信息 (从磁盘读取, 绕过 BufferPool)
  769. func (db *DB) DebugTableState(tableName string) {
  770. tbl, ok := db.tables.Find(tableName)
  771. if !ok {
  772. fmt.Printf("[DEBUG] Table '%s' not found\n", tableName)
  773. return
  774. }
  775. fmt.Printf("[DEBUG] Table '%s': TotalRows=%d (memory)\n", tableName, tbl.info.TotalRows)
  776. // 从磁盘读取 TablePage 检查 TotalRows
  777. buf := make([]byte, PageSize)
  778. if err := db.file.ReadPage(tbl.TablePageID(), buf); err == nil {
  779. info := (*TableInfo)(unsafe.Pointer(&buf[tbl.TableStartOffset()]))
  780. fmt.Printf("[DEBUG] TablePage on disk: TotalRows=%d FieldCount=%d\n", info.TotalRows, info.FieldCount)
  781. }
  782. // 检查索引元数据
  783. im := tbl.IndexMeta()
  784. if im != nil {
  785. for i := range im.metas {
  786. m := &im.metas[i]
  787. if m.Flag&0x80 != 0 {
  788. continue
  789. }
  790. fmt.Printf("[DEBUG] Index '%s': EntryPage=%d IndexType=%d\n",
  791. string(m.IndexName[:]), m.EntryPage, m.IndexType)
  792. // 从磁盘读取 IndexMetaPage 检查 EntryPage
  793. imBuf := make([]byte, PageSize)
  794. if err := db.file.ReadPage(im.PageID(), imBuf); err == nil {
  795. // 读取磁盘上的 EntryPage
  796. diskMeta := (*IndexMeta)(unsafe.Pointer(&imBuf[im.HdrSize()]))
  797. fmt.Printf("[DEBUG] IndexMetaPage on disk: EntryPage=%d\n", diskMeta.EntryPage)
  798. }
  799. // 统计 B+ 树 DataPage 数量
  800. if indexStoresRows(m) {
  801. if handler, ok := index.Compose.GetHandler(m.IndexType); ok {
  802. if rs, ok := handler.(index.RowStoreHandler); ok {
  803. dpIDs := rs.GetDataPageIDs(tbl.handle, m.EntryPage)
  804. totalRows := 0
  805. for _, dpID := range dpIDs {
  806. dpBuf := make([]byte, PageSize)
  807. if err := db.file.ReadPage(dpID, dpBuf); err == nil {
  808. // 读取 DataPage 的行数 (在 offset 18 处的 4 字节 Count)
  809. count := binary.LittleEndian.Uint32(dpBuf[18:])
  810. totalRows += int(count)
  811. }
  812. }
  813. fmt.Printf("[DEBUG] B+ tree: %d DataPages, %d total rows\n", len(dpIDs), totalRows)
  814. }
  815. }
  816. }
  817. }
  818. }
  819. }
  820. // DebugFreePageCount 返回空闲页数量 (调试用)
  821. func (db *DB) DebugFreePageCount() int {
  822. db.file.headerMu.Lock()
  823. n := db.header.FreePageCount
  824. db.file.headerMu.Unlock()
  825. return int(n)
  826. }