| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736 |
- package core
- import (
- "git.ycl.pub/yl32940/TsDB2/DB/core/index"
- "git.ycl.pub/yl32940/TsDB2/DB/core/page"
- "git.ycl.pub/yl32940/TsDB2/DB/core/types"
- "git.ycl.pub/yl32940/TsDB2/DB/core/util"
- "encoding/binary"
- "errors"
- "fmt"
- "strings"
- "sync"
- "sync/atomic"
- "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 表示无事务)。
- // 用 atomic.Pointer: 写线程 (Begin/Commit/Rollback) 与自动提交的定时器
- // goroutine (autoCommit/NotifyWrite) 会并发读写它, 裸指针是 data race。
- // 注意这只保证【指针本身】的可见性 —— Transaction 内部字段仍受
- // "单写入者模型"约束 (同一时刻只允许一个 goroutine 跑写入路径)。
- currentTxn atomic.Pointer[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
- // vacuumMu 串行化 VACUUM 本体: 手动与自动调度可能并发触发,
- // 双开会对同一文件做 Close/Open 交错 (自动调度抢到锁才跑, 否则本轮跳过)。
- vacuumMu sync.Mutex
- // dirtyThreshold 脏页阈值的用户配置 (内存态, wireRuntime 重绑阈值回调时用)。
- dirtyThreshold int
- }
- // 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 {
- // 建库: 未提供介绍 → 头页 Comment 默认写入 VersionComment
- // (当前版本的功能介绍与修复点); 建库时间见 CreateTime 字段
- header, err = newHeader("")
- 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
- }
- // 启动期的句柄用 nil 接收者构造: Init 阶段只会用到 ReadPage/SavePage,
- // 涉及 db 的闭包有 nil 守卫且此阶段不会触发; db 构造完成后以真实接收者重建
- // (wireRuntime), 闭包从此绑定本句柄 —— 见 buildHandleDb 的说明。
- tablePage := &TablePage{
- handle: (*DB)(nil).buildHandleDb(ff, header, realFreePage, realFreeChain),
- }
- // ==========================
- // 初始化索引元数据页
- // ==========================
- var db *DB
- 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,
- dirtyThreshold: DefaultDirtyThreshold,
- }
- // 反向注入, TablePage.Create 时拿到 db 指针; bindDB 顺带给 Init 阶段已经从盘上
- // 加载出来的表补上 db (那时 DB 还没构造完, 只能现在回填)。
- db.tables.bindDB(db)
- db.startAutoCommit()
- // 运行期回调与句柄闭包以真实接收者重绑 (启动期那版绑定的是 nil 接收者)。
- db.wireRuntime()
- return db, nil
- }
- // buildHandleDb 构造 TablePage 的运行期句柄, 闭包绑定接收者 db。
- //
- // Open 在 db 存在之前先用 nil 接收者构造一版 (Init 阶段用不到 db 闭包,
- // 且涉及 db 的闭包都有 nil 守卫); db 构造完成后以真实接收者重建 —— 闭包
- // 从此绑定本句柄的 currentTxn/pendingFree。VACUUM 的 adoptEngineState 换过
- // 引擎状态后也必须用它重绑, 否则闭包仍指向旧句柄: 新句柄上的事务不会被
- // markTouched (Rollback 静默失效), 新分配的页登记到旧句柄 (回滚后成孤儿页)。
- func (db *DB) buildHandleDb(ff *File, header *HeaderPage,
- realFreePage func(uint32), realFreeChain func(uint32, uint32, uint32) error) page.HandleDb {
- return page.HandleDb{
- NewPage: func() uint32 {
- // allocPage(erase=true): headerMu 内摘空闲链 / 递增游标并推进头页; 擦页在
- // 锁外走 WritePage —— 调用方紧接着的整页写入会覆盖池里同一份缓存, 因此不
- // 额外产生物理 IO。读空闲链直接读盘 (绕过 BufferPool)。
- pid := ff.allocPage(header, true)
- // 事务内新分配的页登记到事务: 分配已就地推进 header 游标 (头页延迟到
- // Commit 落盘, 不可回退), Rollback 时必须把这些页还给空闲链, 否则成为
- // 永久泄漏的孤儿页。
- if db != nil {
- if txn := db.currentTxn.Load(); txn != nil {
- txn.allocated = append(txn.allocated, pid)
- }
- }
- return pid
- },
- FreePage: func(pageID uint32) {
- // 事务内被立即还链的页, 从 allocated 撤销登记, 避免 Rollback 时二次归还。
- if db != nil {
- if txn := db.currentTxn.Load(); txn != nil {
- txn.undoAllocated(pageID)
- }
- }
- realFreePage(pageID)
- },
- // 延迟回收: 堆页摘链 / 旧 DataPage 作废时只登记, 等提交成功后再还链。
- // 中途直写盘会让"页已清零"先于 WAL 落盘 —— 崩溃或 Rollback 后原页变空页,
- // 而堆链/索引仍指向它。事务内入 txn.pendingFree (回滚即丢弃),
- // 事务外入 db.pendingFree (下次提交/关闭时统一还链)。
- DeferFreePage: func(pageID uint32) {
- if db == nil {
- realFreePage(pageID) // 启动期(如 WAL 恢复前)没有事务上下文, 保持原行为
- return
- }
- if txn := db.currentTxn.Load(); txn != nil {
- txn.pendingFree = append(txn.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 txn := db.currentTxn.Load(); txn != nil {
- txn.pendingFreeChain = append(txn.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
- },
- }
- }
- // wireRuntime 把运行期回调绑定到本句柄: 事务页登记 (onPageWrite)、脏页阈值
- // 自动提交回调、表页句柄闭包。VACUUM 的 adoptEngineState 换引擎状态后必须重绑,
- // 否则这些闭包仍指向旧句柄。
- func (db *DB) wireRuntime() {
- db.file.SetOnPageWriteCallback(func(pageID uint32) {
- if txn := db.currentTxn.Load(); txn != nil {
- txn.markTouched(pageID)
- }
- })
- // 脏页阈值: 超过时自动触发 Commit 刷盘, 防止内存暴涨。
- // 走 autoCommit: 有活跃事务时不能从中途提交 (会破坏事务原子性)。
- db.file.SetDirtyThreshold(db.dirtyThreshold, db.autoCommit)
- db.tables.handle = db.buildHandleDb(db.file, db.header, db.freePage, db.freeChain)
- }
- // NewPage 分配一个新页号, 走空闲链 / 头页游标的同一路径, 并按事务内已分配登记.
- //
- // 这条路径与 HandleDb.NewPage 闭包完全等价 —— 后者只绑给 TablePage 的几个调用点,
- // 其它地方 (如 BlobCommit) 不持有 handle, 必须走这里. 返回的是页号, 调用方拿到后
- // 仍要自己用 db.file.ReadPage/WritePage 读写内容.
- func (db *DB) NewPage() uint32 {
- pid := db.file.allocPage(db.header, true)
- if txn := db.currentTxn.Load(); txn != nil {
- txn.allocated = append(txn.allocated, pid)
- }
- return pid
- }
- func (db *DB) updateTime() {
- db.file.withHeader(func() {
- db.header.ModifyTime = time.Now().UnixNano()
- _ = db.file.writeHeaderLocked(db.header)
- })
- }
- // Comment 返回头页里的数据库介绍区内容 (建库时自动写入一段创建说明, 之后可用
- // SetComment 改写)。返回的是去掉尾部零字节后的字符串。
- func (db *DB) Comment() string {
- db.file.headerMu.Lock()
- comment := db.header.Comment
- db.file.headerMu.Unlock()
- return strings.TrimRight(string(comment[:]), "\x00")
- }
- // SetComment 改写数据库介绍区 (头页 Comment 字段, ≤512 字节, 超长报错)。
- // 写入走 withHeader + writeHeaderLocked (与 updateTime 同模式), 并更新 ModifyTime。
- func (db *DB) SetComment(desc string) error {
- if len(desc) > len(db.header.Comment) {
- return fmt.Errorf("描述字符串长度 %d 超过最大限制 %d", len(desc), len(db.header.Comment))
- }
- db.file.withHeader(func() {
- var comment [PageSize4K - 332]byte
- copy(comment[:], desc)
- db.header.Comment = comment
- db.header.ModifyTime = time.Now().UnixNano()
- _ = db.file.writeHeaderLocked(db.header)
- })
- return nil
- }
- 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, 提交路径会把头页一并落盘)
- errCommit := db.Commit()
- // 头页兜底: Commit 失败时游标变更还悬着, 必须先于缓冲池刷盘落下,
- // 否则盘上出现页号超过 NextFreePage 的脏页, 重开后会被二次分配覆盖
- errHeader := db.file.FlushHeader(db.header)
- // BufferPool 安全网 (理论上已无脏页)
- db.file.FlushBufferPool()
- // 积压的待回收页: 此时链指针已落盘, 可以安全还链
- db.drainPendingFree()
- // 收尾 fsync: 保证 Header / 数据页真正落到持久介质,
- // 否则进程退出后立刻断电可能丢掉最后一次提交
- errSync := db.file.sync()
- errClose := db.file.Close()
- // 任何一个环节出错都要带出去 (errors.Join 忽略 nil)
- return errors.Join(errCommit, errHeader, errSync, errClose)
- }
- // 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.Load() != 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.Load() != 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.dirtyThreshold = threshold
- 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) {
- if handler, ok := index.Compose.GetHandler(m.IndexType); ok {
- if rs, ok := handler.(index.RowStoreHandler); ok {
- dpIDs := rs.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)
- }
|