db.go 25 KB

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