db.go 26 KB

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