package core import ( "TsDB2/DB/core/index" "TsDB2/DB/core/page" "TsDB2/DB/core/types" "TsDB2/DB/core/util" "encoding/binary" "fmt" "sync" "time" "unsafe" ) type TableMeta struct { Name string FieldCount uint32 TotalRows uint64 FirstHeap uint32 LastHeap uint32 FirstIndex uint32 LastIndex uint32 FirstData uint32 LastData uint32 RowSize uint16 } type DB struct { header *HeaderPage tables *TablePage file *File // path 数据库文件路径 (VACUUM / 自动重打开使用)。 // 由 Open(path) 赋值, 运行时不变。 path string // 防抖自动提交 autoCommitDelay time.Duration autoCommitTimer *time.Timer autoCommitMu sync.Mutex // commitMu 提交互斥: 防抖定时器 / 脏页阈值回调 / 用户显式提交 / 事务提交共用。 // 只串行化"提交"这一步, 不参与写入路径 —— 详见 DB.Commit 的说明。 commitMu sync.Mutex // 当前活跃事务(nil 表示无事务) currentTxn *Transaction // pendingMu 保护 pendingFree / pendingFreeChain: 写入路径登记、提交路径出队, // 两条路径并发 (写线程插入/删除 + 后台提交线程) 时必须串行化, 否则切片并发读写 // 会丢登记项 / 触发数据竞争。 pendingMu sync.Mutex // pendingFree 无事务上下文时积压的"待回收页"(堆页摘链 / 批量删除后作废的旧 DataPage)。 // 不在修改中途直写盘, 而是等 Commit 成功 / Close 后再真正还链 —— 见 DeferFreePage 的说明。 // currentTxn != nil 时改入 txn.pendingFree (Rollback 时整体丢弃)。 pendingFree []uint32 // pendingFreeChain 无事务上下文时积压的"待回收整链"(见 DeferFreeChain)。 // currentTxn != nil 时改入 txn.pendingFreeChain (Rollback 时整体丢弃)。 pendingFreeChain []pendingChain // freePage 真正把页还给空闲链表 (直写盘), 供上面的队列出队时调用 freePage func(pageID uint32) // freeChain 真正把整链还给空闲链表 (直写盘), 供上面的队列出队时调用 freeChain func(head, tail, count uint32) error // 自动 VACUUM 调度 (SetAutoVACUUM 启用) autoVacuumMu sync.Mutex autoVacuumTimer *time.Timer autoVacuumEnabled bool autoVacuumOpts VACUUMOptions lastVACUUM time.Time } // pendingChain 一条待回收的整链: head…tail 共 count 页, 链内 next 在 PageHdr.NextPage type pendingChain struct { head, tail, count uint32 } func Open(path string) (*DB, error) { ff, err := OpenFile(path) if err != nil { return nil, err } // 启用页面缓冲池 ff.InitBufferPool(DefaultBufferPoolCapacity) stat, err := ff.f.Stat() if err != nil { return nil, err } // 自适应缓冲池容量: 按文件总页数 1/4 配置 —— 整库可放内存时最大化命中率, // 巨型库也不会无限制占内存。上界 8000 页 (128MB) 截断, 下界保留默认值。 // 旧固定 16MB 对 214MB+ 的真实库欠配, 1M 行 Delete 全程冷读导致 ~50% 时间 // 花在 pread 上; 扩到 1/4 后, 全表 COUNT(*) / 随机点查的主数据页基本全驻。 if stat.Size() > 0 { filePages := int(stat.Size() / int64(PageSize)) auto := filePages / 4 if auto > 8000 { auto = 8000 } if auto > DefaultBufferPoolCapacity { ff.SetBufferPoolCapacity(auto) } } var header *HeaderPage if stat.Size() == 0 { header, err = newHeader("vrtx") if err != nil { return nil, err } if err := ff.WriteHeader(header); err != nil { return nil, err } } else { header, err = ff.ReadHeader() if err != nil { return nil, err } // 校验魔术字 if header.Magic != Magic { return nil, fmt.Errorf("invalid database file: bad magic %v, expected %v", header.Magic, Magic) } // 校验格式版本: v2 起 schema 改为「精确内联块 + 共享扩展页」, 与 v1 的溢出页链不兼容 if header.Version != Version { return nil, fmt.Errorf("不兼容的数据库文件版本 %d (当前支持 %d), 请重建数据库", header.Version, Version) } } // ========================== // 初始化表页 // ========================== // realFreePage 真正把页还给空闲链表。 // 必须直接写盘: 走 BufferPool 会被 ClearCache 清掉; 且该页在缓存里通常仍是 // "脏"的旧内容, 不覆盖的话 Commit 收集脏页时会把刚写好的空闲链表指针反向 // 覆盖掉, 导致空闲链错乱/页被重复分配。 realFreePage := func(pageID uint32) { ff.withHeader(func() { header.recyclePage(pageID, ff.WritePageDirect) syncBuf := make([]byte, PageSize) _ = ff.readPageDisk(pageID, syncBuf) ff.SyncCache(pageID, syncBuf) _ = ff.writeHeaderLocked(header) }) } // realFreeChain 把一条已用 PageHdr.NextPage 串好的整链一次性还链 (O(1))。 // // 与 realFreePage 的区别只是"链 vs 单页": 这里只改写一条链尾页 + 更新头页, // 代价与链长无关。逐页遍历回收 N 页需要 2N 次 IO (每页一次读 + 一次写), // 拼接只要 3 次 —— DropTable 的堆链、WAL 提交后的 WAL 链都走这里。 realFreeChain := func(head, tail, count uint32) error { var err error ff.withHeader(func() { err = recycleChainWithCache(ff, header, head, tail, count) if err != nil { return } err = ff.writeHeaderLocked(header) }) return err } // 闭包在调用时才解引用 db, 因此可以先声明、稍后再赋值 var db *DB tablePage := &TablePage{ handle: page.HandleDb{ NewPage: func() uint32 { // allocPage(erase=true): headerMu 内摘空闲链 / 递增游标并推进头页; 擦页在 // 锁外走 WritePage —— 调用方紧接着的整页写入会覆盖池里同一份缓存, 因此不 // 额外产生物理 IO。读空闲链直接读盘 (绕过 BufferPool)。 pid := ff.allocPage(header, true) // 事务内新分配的页登记到事务: 分配已就地推进 header 游标 (头页延迟到 // Commit 落盘, 不可回退), Rollback 时必须把这些页还给空闲链, 否则成为 // 永久泄漏的孤儿页。 if db != nil && db.currentTxn != nil { db.currentTxn.allocated = append(db.currentTxn.allocated, pid) } return pid }, FreePage: func(pageID uint32) { // 事务内被立即还链的页, 从 allocated 撤销登记, 避免 Rollback 时二次归还。 if db != nil && db.currentTxn != nil { db.currentTxn.undoAllocated(pageID) } realFreePage(pageID) }, // 延迟回收: 堆页摘链 / 旧 DataPage 作废时只登记, 等提交成功后再还链。 // 中途直写盘会让"页已清零"先于 WAL 落盘 —— 崩溃或 Rollback 后原页变空页, // 而堆链/索引仍指向它。事务内入 txn.pendingFree (回滚即丢弃), // 事务外入 db.pendingFree (下次提交/关闭时统一还链)。 DeferFreePage: func(pageID uint32) { if db == nil { realFreePage(pageID) // 启动期(如 WAL 恢复前)没有事务上下文, 保持原行为 return } if db.currentTxn != nil { db.currentTxn.pendingFree = append(db.currentTxn.pendingFree, pageID) return } db.pendingMu.Lock() db.pendingFree = append(db.pendingFree, pageID) db.pendingMu.Unlock() }, ReadPage: ff.ReadPage, SavePage: ff.WritePage, // 整链回收: DropTable 的堆链 / WAL 提交后的 WAL 链 / 哈希索引的桶目录链与桶 // 溢出链, 都是「首尾和链长已知、且链内 next 与空闲链同用 PageHdr.NextPage」 // 的链 —— 一次拼接 O(1) 还链, 不必逐页遍历 (逐页 = 每页读+写各一次)。 FreeChain: realFreeChain, // 延迟整链回收: 与 DeferFreePage 同一时序约定 (提交成功后才真正还链), // 但一条链只登记一次, 提交后一次 FreeChain 还链。 DeferFreeChain: func(head, tail, count uint32) { if head == 0 || count == 0 { return } if db == nil { _ = realFreeChain(head, tail, count) // 启动期没有事务上下文, 保持原行为 return } c := pendingChain{head: head, tail: tail, count: count} if db.currentTxn != nil { db.currentTxn.pendingFreeChain = append(db.currentTxn.pendingFreeChain, c) return } db.pendingMu.Lock() db.pendingFreeChain = append(db.pendingFreeChain, c) db.pendingMu.Unlock() }, // ValidPage: 页号合法性 = 落在已分配区间 [2, NextFreePage)。 // 0 是头页、1 是建库预留的表页; >= NextFreePage 的页从未分配过, 文件里没有 // 它的位置 —— 放进空闲链, 下一次 NewPage 就会拿着它写到几十 GB 偏移处 // (实测 DROP TABLE 假死 52s、稀疏文件暴涨到 40GB 的直接原因)。 ValidPage: func(pageID uint32) bool { ff.headerMu.Lock() limit := header.NextFreePage ff.headerMu.Unlock() return pageID >= 2 && pageID < limit }, }, } // ========================== // 初始化索引元数据页 // ========================== saveHeader := false //如果索引页和表页都是空的话则直接创建 if header.TablePage == 0 { // 用带读盘能力的分配: 即便此时空闲链非空 (例如 TablePage 丢失的旧库), // 也能正确从空闲链摘页; 新库空闲链为空, 走 NextFreePage 递增分配。 header.TablePage = ff.allocPage(header, true) tablePage.InitHeader(header.TablePage) saveHeader = true } tablePage.Init(header.TablePage) if saveHeader { err = ff.WriteHeader(header) if err != nil { return nil, err } } // ========================== // WAL 恢复: 检查是否有未完成的 WAL 事务 // ========================== if err := RecoverWAL(ff, header); err != nil { return nil, err } db = &DB{ file: ff, header: header, tables: tablePage, path: path, autoCommitDelay: DefaultAutoCommitDelay, freePage: realFreePage, freeChain: realFreeChain, } // 反向注入, TablePage.Create 时拿到 db 指针; bindDB 顺带给 Init 阶段已经从盘上 // 加载出来的表补上 db (那时 DB 还没构造完, 只能现在回填)。 db.tables.bindDB(db) db.startAutoCommit() // 事务回调: BufferPool.WritePage 时登记本事务修改过的页, // 供 Rollback 精确丢弃这些页的缓存 (不影响事务开始前遗留的脏页) ff.SetOnPageWriteCallback(func(pageID uint32) { if db.currentTxn != nil { db.currentTxn.markTouched(pageID) } }) // 脏页阈值: 超过时自动触发 Commit 刷盘, 防止内存暴涨 // 走 autoCommit: 有活跃事务时不能从中途提交 (会破坏事务原子性) ff.SetDirtyThreshold(DefaultDirtyThreshold, db.autoCommit) return db, nil } // NewPage 分配一个新页号, 走空闲链 / 头页游标的同一路径, 并按事务内已分配登记. // // 这条路径与 HandleDb.NewPage 闭包完全等价 —— 后者只绑给 TablePage 的几个调用点, // 其它地方 (如 BlobCommit) 不持有 handle, 必须走这里. 返回的是页号, 调用方拿到后 // 仍要自己用 db.file.ReadPage/WritePage 读写内容. func (db *DB) NewPage() uint32 { pid := db.file.allocPage(db.header, true) if db.currentTxn != nil { db.currentTxn.allocated = append(db.currentTxn.allocated, pid) } return pid } func (db *DB) updateTime() { db.file.withHeader(func() { db.header.ModifyTime = time.Now().UnixNano() _ = db.file.writeHeaderLocked(db.header) }) } func (db *DB) FindTable(name string) (*Table, bool) { tbl, ok := db.tables.Find(name) if ok && tbl.onWrite == nil { tbl.SetOnWrite(db.NotifyWrite) } return tbl, ok } func (db *DB) CreateTable(name string, fields []types.FieldInfo) (*Table, error) { tbl, err := db.tables.Create(name, fields) if err != nil { return nil, err } tbl.SetOnWrite(db.NotifyWrite) db.file.withHeader(func() { db.header.TableCount++ _ = db.file.writeHeaderLocked(db.header) }) db.updateTime() return tbl, nil } func (db *DB) DropTable(name string) error { // 先刷脏页再清缓存: ClearCache 会把缓存里的页统统标成"干净", // 而随后的 FlushBufferPool(CollectDirty) 会把它们直接移出缓存 —— // 如果缓存里还有尚未落盘的页(手动提交模式下很常见), 就会被静默丢掉。 db.file.FlushBufferPool() db.file.ClearCache() if err := db.tables.Drop(name); err != nil { return err } db.file.withHeader(func() { db.header.TableCount-- _ = db.file.writeHeaderLocked(db.header) }) db.updateTime() // 先刷盘再清缓存, 防止 Drop 中标记的 Flag(0x80) 丢失 db.file.FlushBufferPool() db.file.ClearCache() return nil } func (db *DB) GetAllTables() []TableMeta { res := make([]TableMeta, 0, len(db.tables.tables)) for _, t := range db.tables.tables { if util.IsHighBitSet(t.info.Flag) { continue } res = append(res, TableMeta{ Name: util.BytesToString(t.info.Name[:]), FieldCount: uint32(t.info.FieldCount), TotalRows: t.info.TotalRows, FirstHeap: t.info.FirstHeap, LastHeap: t.info.LastHeap, FirstData: t.info.FirstData, LastData: t.info.LastData, RowSize: t.info.RowSize, }) } return res } func (db *DB) Close() error { // 停止防抖定时器 db.stopAutoCommit() // 先提交剩余脏页 (走 WAL, 提交路径会把头页一并落盘) _ = db.Commit() // 头页兜底: Commit 失败时游标变更还悬着, 必须先于缓冲池刷盘落下, // 否则盘上出现页号超过 NextFreePage 的脏页, 重开后会被二次分配覆盖 _ = db.file.FlushHeader(db.header) // BufferPool 安全网 (理论上已无脏页) db.file.FlushBufferPool() // 积压的待回收页: 此时链指针已落盘, 可以安全还链 db.drainPendingFree() // 收尾 fsync: 保证 Header / 数据页真正落到持久介质, // 否则进程退出后立刻断电可能丢掉最后一次提交 _ = db.file.sync() return db.file.Close() } // freeNow 把一批待回收页真正还给空闲链表 (直写盘) func (db *DB) freeNow(ids []uint32) { for _, pid := range ids { if db.freePage != nil { db.freePage(pid) } } } // freeChainNow 把一批待回收整链真正还给空闲链表 (每条链一次 O(1) 拼接) func (db *DB) freeChainNow(chains []pendingChain) { for _, c := range chains { if db.freeChain != nil { _ = db.freeChain(c.head, c.tail, c.count) } } } // drainPendingFree 回收无事务上下文时积压的页与整链。 // 只能在"链指针已经持久化"的时刻调用 —— 即提交成功之后, 或 Close 收尾时。 func (db *DB) drainPendingFree() { // 先原子地摘下当前队列 (写入路径可能正在并发登记, 只取走此刻已入队的部分), // 再在锁外真正还链 —— 还链会走 headerMu + 磁盘 IO, 不该占着 pendingMu。 db.pendingMu.Lock() ids := db.pendingFree db.pendingFree = nil chains := db.pendingFreeChain db.pendingFreeChain = nil db.pendingMu.Unlock() if len(ids) == 0 && len(chains) == 0 { return } db.freeNow(ids) db.freeChainNow(chains) } // Commit 显式提交 — 通过 WAL 保证原子性 // 同时重置防抖定时器 // // commitMu 保证同一时刻只有一笔提交在飞: 防抖定时器、脏页阈值回调(在写入 goroutine // 内同步触发) 和用户显式调用都可能撞在一起, 而两笔提交同时收集/回写同一批脏页会把 // 页的归属搞乱 (前一笔的快照会把后一笔刚改出来的内容判成"已落盘")。 // 这把锁只串行化提交本身, 不参与写入路径: 写入只在 BufferPool.WritePage 里等 // writeLock (见 LockWrites), 两者没有嵌套获取, 因此不会互相死锁。 func (db *DB) Commit() error { db.commitMu.Lock() defer db.commitMu.Unlock() // 取消待执行的自动提交 db.autoCommitMu.Lock() if db.autoCommitTimer != nil { db.autoCommitTimer.Stop() } db.autoCommitMu.Unlock() // 1. 收集所有 BufferPool 脏页 (TablePage + IndexPage + DataPage 等) dirty := db.file.CollectDirty() if len(dirty) == 0 { // 无脏页 = 修改已持久, 积压的待回收页可以安全还链 db.drainPendingFree() return nil // 无脏页, 跳过 } // 2. 创建事务 wal := db.BeginTxn() // 3. 记录所有脏页 for _, d := range dirty { wal.RecordPage(d.PageID, d.Data) } // 4. HeaderPage 也纳入 (pageID=0) // marshalHeader 要读整张头页 (含提交路径正在改的 WAL 链指针), 必须在 headerMu 内拍快照。 db.file.withHeader(func() { wal.RecordPage(0, marshalHeader(db.header)) }) // 5. 原子提交 if err := wal.Commit(); err != nil { // 提交失败: 这批页必须继续留在缓存里当脏页。CollectDirty 现在不再把它们移出 // 缓存, 所以正常情况下它们本来就在; RestoreDirty 只是兜底 (错误路径上页回收 // 等动作也可能碰过这些页), 而且只补缺失条目, 不会用旧快照盖掉更新的内容。 db.file.RestoreDirty(dirty) return err } // 6. 提交成功 → 把这一批页收成干净页。 // 必须按快照的版本号逐页判定: 提交窗口里写入线程可能已经把某些页又改了一遍 // (缓存里是更新的内容, 本次 WAL 记的却是快照那份), 那些页要继续留脏, 否则新 // 内容既不会被写盘, 脏标记还报告"已持久化" —— 该页一旦被淘汰就永久丢失。 // (旧实现这里是 ClearAll, 无差别地把整个缓存标干净, 恰好会抹掉这种页。) db.file.MarkClean(dirty) // 7. 提交成功 → 链指针已持久化, 积压的待回收页现在可以真正还链 db.drainPendingFree() return nil } // ======================== 防抖自动提交 ======================== // SetAutoCommitDelay 设置防抖自动提交延迟 // delay=0 表示禁用自动提交 (仅手动 Commit) func (db *DB) SetAutoCommitDelay(d time.Duration) { db.autoCommitMu.Lock() defer db.autoCommitMu.Unlock() db.autoCommitDelay = d if db.autoCommitTimer != nil { db.autoCommitTimer.Stop() db.autoCommitTimer = nil } } // NotifyWrite 写操作通知 (InsertRow/DeleteRow/UpdateRow 完成后调用) // 重置防抖定时器: 如果延迟期间没有新操作, 则自动 Commit func (db *DB) NotifyWrite() { if db.autoCommitDelay <= 0 { return } // 有活跃事务时不启动自动提交: 否则定时器到点会把事务从中途截断提交, // 破坏事务原子性, 并使后续 Rollback 无法真正撤销 if db.currentTxn != nil { return } db.autoCommitMu.Lock() defer db.autoCommitMu.Unlock() // 复用同一个 timer 对象,避免频繁创建/销毁 if db.autoCommitTimer == nil { // 首次创建 db.autoCommitTimer = time.AfterFunc(db.autoCommitDelay, db.autoCommit) } else { // 后续重置(无需重新分配对象,减少 GC 压力) db.autoCommitTimer.Reset(db.autoCommitDelay) } } // autoCommit 自动提交入口 (防抖定时器 / 脏页阈值共用) // 有活跃事务时直接跳过, 避免把事务从中途截断提交 func (db *DB) autoCommit() { if db.currentTxn != nil { return } _ = db.Commit() } // startAutoCommit 启动防抖自动提交 func (db *DB) startAutoCommit() { // 不立即启动定时器, 等待第一次 NotifyWrite } // stopAutoCommit 停止防抖自动提交 func (db *DB) stopAutoCommit() { db.autoCommitMu.Lock() defer db.autoCommitMu.Unlock() if db.autoCommitTimer != nil { db.autoCommitTimer.Stop() db.autoCommitTimer = nil } } // SetDirtyThreshold 设置脏页阈值(页数),超过此值自动触发 Commit func (db *DB) SetDirtyThreshold(threshold int) { db.file.SetDirtyThreshold(threshold, db.autoCommit) } // LockWrites 锁定写入 —— 实际锁的是"页落盘"(BufferPool.WritePage 会 waitForWriteLock)。 // 锁定期 Insert/Update/Delete 仍可改内存中的页并标记脏, 只是要写磁盘时阻塞; 没有 // "写入被拒绝"的错误码。Commit/Flush 期间由引擎内部调用, 以取得稳定的磁盘视图。 func (db *DB) LockWrites() { db.file.LockWrites() } // UnlockWrites 解锁写入 (见 LockWrites 的语义说明) func (db *DB) UnlockWrites() { db.file.UnlockWrites() } // SetBufferPoolCapacity 设置 BufferPool 容量(页数) func (db *DB) SetBufferPoolCapacity(capacity int) { db.file.SetBufferPoolCapacity(capacity) } // marshalHeader 将 HeaderPage 序列化为字节切片 (填充到 PageSize) func marshalHeader(h *HeaderPage) []byte { buf := make([]byte, PageSize) src := (*[PageSize4K]byte)(unsafe.Pointer(h))[:] copy(buf, src) return buf } // ======================== 调试诊断 ======================== // DebugDBState 打印数据库调试信息 func (db *DB) DebugDBState() { db.file.withHeader(func() { fmt.Printf("[DEBUG] Header: TablePage=%d NextFreePage=%d FreePageStart=%d FirstWAL=%d LastWAL=%d NextTxn=%d\n", db.header.TablePage, db.header.NextFreePage, db.header.FreePageStart, db.header.FirstWALPage, db.header.LastWALPage, db.header.NextTxnID) }) fmt.Printf("[DEBUG] BufferPool: total=%d dirty=%d\n", db.file.bp.Len(), db.file.bp.DirtyCount()) } // DebugTableState 打印表调试信息 (从磁盘读取, 绕过 BufferPool) func (db *DB) DebugTableState(tableName string) { tbl, ok := db.tables.Find(tableName) if !ok { fmt.Printf("[DEBUG] Table '%s' not found\n", tableName) return } fmt.Printf("[DEBUG] Table '%s': TotalRows=%d (memory)\n", tableName, tbl.info.TotalRows) // 从磁盘读取 TablePage 检查 TotalRows buf := make([]byte, PageSize) if err := db.file.ReadPage(tbl.TablePageID(), buf); err == nil { info := (*TableInfo)(unsafe.Pointer(&buf[tbl.TableStartOffset()])) fmt.Printf("[DEBUG] TablePage on disk: TotalRows=%d FieldCount=%d\n", info.TotalRows, info.FieldCount) } // 检查索引元数据 im := tbl.IndexMeta() if im != nil { for i := range im.metas { m := &im.metas[i] if m.Flag&0x80 != 0 { continue } fmt.Printf("[DEBUG] Index '%s': EntryPage=%d IndexType=%d\n", string(m.IndexName[:]), m.EntryPage, m.IndexType) // 从磁盘读取 IndexMetaPage 检查 EntryPage imBuf := make([]byte, PageSize) if err := db.file.ReadPage(im.PageID(), imBuf); err == nil { // 读取磁盘上的 EntryPage diskMeta := (*IndexMeta)(unsafe.Pointer(&imBuf[im.HdrSize()])) fmt.Printf("[DEBUG] IndexMetaPage on disk: EntryPage=%d\n", diskMeta.EntryPage) } // 统计 B+ 树 DataPage 数量 if indexStoresRows(m) { handler, _ := index.Compose.GetHandler(m.IndexType) if handler != nil { dpIDs := handler.GetDataPageIDs(tbl.handle, m.EntryPage) totalRows := 0 for _, dpID := range dpIDs { dpBuf := make([]byte, PageSize) if err := db.file.ReadPage(dpID, dpBuf); err == nil { // 读取 DataPage 的行数 (在 offset 18 处的 4 字节 Count) count := binary.LittleEndian.Uint32(dpBuf[18:]) totalRows += int(count) } } fmt.Printf("[DEBUG] B+ tree: %d DataPages, %d total rows\n", len(dpIDs), totalRows) } } } } } // DebugFreePageCount 返回空闲页数量 (调试用) func (db *DB) DebugFreePageCount() int { db.file.headerMu.Lock() n := db.header.FreePageCount db.file.headerMu.Unlock() return int(n) }