package core import ( "encoding/binary" "errors" "fmt" "strings" "sync" "sync/atomic" "time" "unsafe" "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" ) 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 // accMu 操作级可重入串行锁: 引擎不支持并发读写, 所有公开操作入口都过它, // 把多 goroutine 的误并发自动串行化(防止改坏结构 / 读到中间态)。 // 显式事务在 Begin 持有、Commit/Rollback 释放; 事务内同 goroutine 重入不阻塞。 // 详见 access_lock.go。 accMu accessLock // 当前活跃事务(nil 表示无事务)。 // 用 atomic.Pointer: 写线程 (Begin/Commit/Rollback) 与自动提交的定时器 // goroutine (autoCommit/NotifyWrite) 会并发读写它, 裸指针是 data race。 // 注意这只保证【指针本身】的可见性 —— Transaction 内部字段仍受 // "单写入者模型"约束 (同一时刻只允许一个 goroutine 跑写入路径)。 currentTxn atomic.Pointer[Transaction] // opDepth 行级多页写操作的嵌套深度 (InsertRow/UpdateRow/DeleteRow/批量/建索引/批量删)。 // >0 时阈值/防抖提交一律不触发: 否则 Commit 会把同一操作的脏页切成两笔 WAL, // kill 后烧出半成品 (DockerManager 实测: 数据页已提交、索引页未提交, 行按主键 // 在、按二级索引永远查不到)。深度归零时由收尾闭包补一次提交检查。 opDepth atomic.Int32 // 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 } // OpenOptions 打开数据库的选项。 type OpenOptions struct { // ForceInspect 抢救模式: 头页校验 (Magic/XXH32) 失败时仍打开 —— 打印警告、 // 只读 (一切写操作返回 ErrReadOnly)、跳过 WAL 恢复 (恢复会写盘)。 // 用途: 头页损坏后抢救数据 / analyzer 类工具手工勘验。 ForceInspect bool // ReadOnly 只读打开: 文件以 O_RDONLY 打开、加【共享】跨进程锁 (可与其它只读 // 进程共存, 与读写进程互斥)、跳过 WAL 恢复与一切写盘。用于报表 / 分析等不 // 修改数据的连接 (DSN mode=ro)。 ReadOnly bool } // Open 打开数据库 (等价 OpenWithOptions 零值)。 func Open(path string) (*DB, error) { return OpenWithOptions(path, OpenOptions{}) } // OpenWithOptions 打开数据库。新建库走 newHeader + PrepareWrite (Version=11, // CRC 随首次落盘生成); 已有库先 Validate (Magic 必查, v11 且 CRC==0 的旧库 // 跳过, 其余校验 XXH32), 校验失败默认拒绝打开, ForceInspect 时只读抢救。 func OpenWithOptions(path string, opt OpenOptions) (*DB, error) { var ( ff *File err error ) if opt.ReadOnly { ff, err = OpenFileReadOnly(path) } else { ff, err = OpenFile(path) } if err != nil { return nil, err } // 启用页面缓冲池 (只读时仅作读缓存, 不会产生脏页) ff.InitBufferPool(DefaultBufferPoolCapacity) stat, err := ff.f.Stat() if err != nil { return nil, err } if opt.ReadOnly && stat.Size() == 0 { _ = ff.f.Close() return nil, fmt.Errorf("只读打开失败: 数据库文件 %q 不存在或为空", path) } // 自适应缓冲池容量: 按文件总页数 1/4 配置 —— 整库可放内存时最大化命中率, // 巨型库也不会无限制占内存。上界 MaxBufferPoolCapacity 页截断, 下界保留默认值。 // 旧固定 16MB 对 214MB+ 的真实库欠配, 1M 行 Delete 全程冷读导致 ~50% 时间 // 花在 pread 上; 扩到 1/4 后, 全表 COUNT(*) / 随机点查的主数据页基本全驻。 if stat.Size() > 0 { filePages := int(stat.Size() / int64(PageSize)) auto := filePages / 4 if auto > MaxBufferPoolCapacity { auto = MaxBufferPoolCapacity } if auto > DefaultBufferPoolCapacity { ff.SetBufferPoolCapacity(auto) } } // 跨进程文件锁: 只读用共享锁 (多个只读可共存), 读写用排他锁; 在读头页 / // WAL 恢复之前先占锁, 阻止读写进程与读者/写者同时打开。 if stat.Size() > 0 { var lockErr error if opt.ReadOnly { lockErr = ff.tryRLock() } else { lockErr = ff.tryLock() } if lockErr != nil { _ = ff.f.Close() // 释放句柄 (锁随之释放) return nil, lockErr } } 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 } // 新库头页写入后文件已有内容, 立即加跨进程文件锁。 if err := ff.tryLock(); err != nil { _ = ff.f.Close() return nil, err } } else { header, err = ff.ReadHeader() if err != nil { return nil, err } // 头页校验: Magic 必查; CRC 对 v11-旧库 (HeaderCrc==0) 跳过, 其余校验。 if err := header.Validate(); err != nil { // 只读打开本就不写; 抢救模式显式要求 —— 两者都降级为只读, 其余情况拒绝。 if !opt.ForceInspect && !opt.ReadOnly { _ = ff.f.Close() // 校验失败必须释放文件句柄, 否则 Windows 上文件被占用 return nil, err } fmt.Printf("警告: %v\n以只读模式打开 (写操作被禁用)\n", err) ff.readOnly = true } // 版本闸门: 低于当前版本的库结构对不上, 一律拒绝 (抢救模式也不例外) if header.Version < Version { _ = ff.f.Close() 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 { if opt.ReadOnly { _ = ff.f.Close() return nil, fmt.Errorf("只读打开的数据库缺少表页 (TablePage=0): 不是有效数据库, 无法在只读模式初始化") } // 用带读盘能力的分配: 即便此时空闲链非空 (例如 TablePage 丢失的旧库), // 也能正确从空闲链摘页; 新库空闲链为空, 走 NextFreePage 递增分配。 header.TablePage = ff.allocPage(header, true) tablePage.InitHeader(header.TablePage) saveHeader = true } if saveHeader { err = ff.WriteHeader(header) if err != nil { return nil, err } } // ========================== // WAL 恢复: 检查是否有未完成的 WAL 事务 // ========================== // WAL 恢复会写盘 —— 抢救模式 (只读) 下跳过: 待恢复事务保持挂起, 数据页 // 仍是盘上原状, 读路径不受影响 (这正是抢救模式要的效果)。 // // 必须在 tablePage.Init 之前: 恢复会把已提交事务的表信息页镜像 (TotalRows / // 堆链指针 / RowSize 等) 重放回磁盘, 而 Init 会把这些元数据【载入内存 Table // 对象】。先 Init 再恢复, 内存里就是旧元数据: 崩溃恢复后 TotalRows 对不上、 // 堆链指针过期, 后续写入按过期结构分配 = 损坏 (wal_crash_chaos 压测实锤)。 if !ff.readOnly { if err := RecoverWAL(ff, header); err != nil { _ = ff.f.Close() // 恢复失败必须释放文件句柄, 否则 Windows 上文件被占用 return nil, err } // 孤儿 WAL 链自愈: kill 恰落在"WAL 页已写盘、头页指针未发布(或已被后续 // 提交清掉)"的窗口时, 盘上会留下 FirstWALPage 找不到的 WAL 链页, 永不 // 回收。这里把它们整条捡回空闲链 (不重放: 已提交事务的回写必先于指针 // 清零落盘, 重放只是幂等; 未提交事务的镜像本就该丢弃 —— 回收即丢弃)。 if err := healOrphanWALPages(ff, header); err != nil { _ = ff.f.Close() return nil, err } } // 恢复完成后再载入表元数据: 上面重放过的表页此刻才是最新状态 tablePage.Init(header.TablePage) 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 }, } } // errIfReadOnly 抢救模式 (force-inspect) 下写入入口的统一闸门。 // File 层的 readOnly 守卫保证"任何数据都不可能落盘"; 这里的入口闸门让写入 // 在第一时间就拿到清晰的 ErrReadOnly, 而不是缓存一串脏页到提交时才失败。 func (db *DB) errIfReadOnly() error { if db.file.readOnly { return ErrReadOnly } return nil } // errIfReadOnly 表级写入入口的闸门 (经 table.db 转发)。 func (table *Table) errIfReadOnly() error { if table.db == nil { return nil } return table.db.errIfReadOnly() } // 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 err := db.errIfReadOnly(); err != nil { return err } defer db.lockOp()() if len(desc) > len(db.header.Comment) { return fmt.Errorf("描述字符串长度 %d 超过最大限制 %d", len(desc), len(db.header.Comment)) } var werr error db.file.withHeader(func() { var comment [PageSize4K - 332]byte copy(comment[:], desc) db.header.Comment = comment db.header.ModifyTime = time.Now().UnixNano() werr = db.file.writeHeaderLocked(db.header) }) return werr } 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) { if db.HasActiveTransaction() { return nil, errDDLInTransaction("CreateTable") } defer db.lockOp()() // DDL 全程串行; 内部 db.Commit 同 goroutine 重入 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() // DDL 隐式提交: 把新表钉到盘上 —— DDL 不具有事务性 (与 Oracle/MySQL 一致), // 未提交的 DDL 随回滚消失后, 内存元数据 (TableInfo/IndexMeta) 会指向已回 // 收的页, 后续读写立即迷路 (txn_crash_chaos 压测实锤) if err := db.Commit(); err != nil { return nil, err } return tbl, nil } // errDDLInTransaction 事务内执行 DDL 的明确拒绝。 // // DDL 一旦允许进入事务, 回滚它就会出现"内存元数据 vs 磁盘真值"的分叉: // Rollback 按页丢弃缓存, 而 Table.info / IndexMeta.metas 是独立于页缓存的 // Go 堆副本, 不会随缓存一起消失 —— 未提交的建表/建索引回滚后, 它们仍指向 // 事务中分配、已被回收的页号, 下一个操作立刻 load failed / 写穿到空闲页。 // 因此这里宁可显式拒绝, 也不让调用方踩进这个不可恢复的坑。 func errDDLInTransaction(op string) error { return fmt.Errorf("%s: DDL 操作不支持在事务中执行 (DDL 为隐式提交语义, 请先 Commit 或 Rollback 当前事务)", op) } // ddlBoundary DDL 收尾: 无活跃事务时隐式提交, 有活跃事务时应已被入口拒绝 // (防御性再查一次)。返回 error 时 DDL 已成功但落盘失败 —— 调用方应视 DDL 为未完成。 func (db *DB) ddlBoundary(op string) error { if db.HasActiveTransaction() { return errDDLInTransaction(op) } return db.Commit() } func (db *DB) DropTable(name string) error { if db.HasActiveTransaction() { return errDDLInTransaction("DropTable") } defer db.lockOp()() // DDL 全程串行; 内部 Flush/Commit 同 goroutine 重入 // 先刷脏页再清缓存: 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 db.ddlBoundary("DropTable") } // TruncateTable 清空一张表的全部数据但保留表结构与索引定义 (DDL, 隐式提交)。 // 事务内拒绝 (与其它 DDL 一致)。 func (db *DB) TruncateTable(name string) error { if err := db.errIfReadOnly(); err != nil { return err } if db.HasActiveTransaction() { return errDDLInTransaction("TruncateTable") } defer db.lockOp()() if err := db.tables.Truncate(name); err != nil { return err } return db.ddlBoundary("TruncateTable") } // RenameTable 重命名表 (DDL, 隐式提交)。只改 TableInfo.Name 元数据字符串, // 不涉及数据页 / 索引页迁移或重建。事务内拒绝。 func (db *DB) RenameTable(oldName, newName string) error { if err := db.errIfReadOnly(); err != nil { return err } if db.HasActiveTransaction() { return errDDLInTransaction("RenameTable") } defer db.lockOp()() if err := db.tables.RenameTable(oldName, newName); err != nil { return err } return db.ddlBoundary("RenameTable") } // RenameColumn 重命名列 (DDL, 隐式提交)。只改 FieldInfo.FieldName + schema 落盘, // 不涉及列偏移 / 行数据 / 索引键重建。事务内拒绝。 func (db *DB) RenameColumn(tableName, oldCol, newCol string) error { if err := db.errIfReadOnly(); err != nil { return err } if db.HasActiveTransaction() { return errDDLInTransaction("RenameColumn") } defer db.lockOp()() tbl, ok := db.tables.Find(tableName) if !ok { return ErrTableNotFound } return tbl.RenameColumn(oldCol, newCol) } 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 { defer db.lockOp()() // 等待在飞操作结束, 并在关闭全程挡住新操作 // 停止防抖定时器 db.stopAutoCommit() // 只读实例: 无脏页/无待写, 直接关句柄 —— Commit/写头/fsync 在只读 fd 上 // (Windows) 会报 Access is denied, 且没有任何需要持久化的内容。 if db.file.readOnly { return db.file.Close() } // 先提交剩余脏页 (走 WAL, 提交路径会把头页一并落盘) errCommit := db.Commit() // 头页兜底: Commit 失败时游标变更还悬着, 必须先于缓冲池刷盘落下, // 否则盘上出现页号超过 NextFreePage 的脏页, 重开后会被二次分配覆盖 errHeader := db.file.FlushHeader(db.header) // BufferPool 安全网 (理论上已无脏页); 写盘失败必须带出去 —— 失败页仍脏, // 但 Close 之后无人再刷, 吞掉错误等于静默丢数据 errFlush := db.file.FlushBufferPool() // 积压的待回收页: 此时链指针已落盘, 可以安全还链 db.drainPendingFree() // 收尾 fsync: 保证 Header / 数据页真正落到持久介质, // 否则进程退出后立刻断电可能丢掉最后一次提交 errSync := db.file.sync() errClose := db.file.Close() // 任何一个环节出错都要带出去 (errors.Join 忽略 nil) return errors.Join(errCommit, errHeader, errFlush, 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 { if err := db.errIfReadOnly(); err != nil { return err } defer db.lockOp()() // 操作级串行: 提交期间挡住其他 goroutine 的读写(独立调用时) 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): allocPage 推进分配游标 // 会置 headerDirty, 新页号必须随本笔 WAL 落盘 —— 否则崩溃重放后盘上旧游标 // 会把这批新页二次分配。纯改写已有页的提交不动头页, 省掉 4K 镜像与回写; // NextTxnID / ModifyTime 这类诊断字段只随内存游标前进, 不强制头页落盘。 if db.file.IsHeaderDirty() { // 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) } } // beginOp 标记一个行级多页写操作的开始, 返回闭包在操作结束时调用 (defer 形态, // 嵌套安全: 深度计数, 最外层收尾)。收尾只做一件事: 重新武装防抖提交 —— 提交 // 时机仍由防抖/阈值策略决定, 这里保证的是"任何提交都不早于操作结束"。 func (db *DB) beginOp() func() { db.opDepth.Add(1) return func() { if db.opDepth.Add(-1) == 0 { if db.file.DirtyOverThreshold() { db.autoCommit() // 阈值在操作期间被越过 (被栅栏压住): 立即补提交 } else { db.NotifyWrite() // 否则重新武装防抖定时器 } } } } // autoCommit 自动提交入口 (防抖定时器 / 脏页阈值共用) // 有活跃事务时直接跳过, 避免把事务从中途截断提交; 行级写操作中途同样跳过 // (opDepth>0), 等操作收尾时由 beginOp 的闭包统一补提交。 func (db *DB) autoCommit() { if db.currentTxn.Load() != nil { return } if db.opDepth.Load() > 0 { 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) // 在【副本】上重算 CRC 再快照: 内存中的 HeaderCrc 可能滞后于游标 / 链 // 指针的内存内变更 (allocPage 只标脏不落盘), 快照字节必须与快照内容自洽, // 否则崩溃恢复重放出头页后 CRC 与内容对不上。 snap := *h snap.PrepareWrite() src := (*[PageSize4K]byte)(unsafe.Pointer(&snap))[:] 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) }