db.go 23 KB

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