package core import ( "TsDB2/DB/core/page" "encoding/binary" "hash/crc32" ) // ============================================================================= // WAL (Write-Ahead Log) 实现 // // WAL 页布局 (16KB), 权威定义见 page.WALPageHdr: // [WALPageHdr(40B)] [WALEntry...] // 首页与续页头部等长 —— 续页的 TxnID/TotalEntries/State 填 0, 换来整条链能用 // 同一个结构体解析, 读写都不再按 isFirst 分叉。 // // ⭐ 页间链指针用 PageHdr.NextPage (与空闲链同一个字段), 不再自定义 NextWalPage —— // 这样整条 WAL 链本身就是一条合法空闲链, 回收时一次拼接即可, 不必逐页遍历 (见 recycleChain)。 // // WALEntry (见 page.WALEntryMeta): [TargetPageID(4B)] [DataLen(4B)] [Checksum(4B)] [OpType(1B)] [Data(DataLen)] // OpType == WALOpFullPage(0) → Data 为整页镜像 (DataLen = 页有效长度) // OpType == Insert/Delete/Update → Data 为行级数据 // Checksum = CRC32(OpType || Data) // // ⚠️ OpType 必须显式存储: 早期实现靠"Data 的首字节"猜类型, 但整页镜像的首字节 // 就是 PageHdr.Type, 而 DataPage/TablePage/HeapPage 的 Type 恰好是 2/1/3, // 与行级 OpType 冲突, 整页记录会被误判成行级记录而丢失。 // // WAL 状态机 (State 字段): // 0 = Mem (内存中,未写盘) // 1 = Written (已 write(),未 fsync()) // 2 = Persisted (已 fsync(),但未标记提交) // 3 = Committed (已提交,可安全恢复) // // 打包: 一个 WAL 页可容纳**多条** Entry —— 写端按「当前页剩余空间 >= 13B + DataLen」 // 紧凑追加, 放不下才开新页。整页镜像 (≈16KB) 仍会独占一页; 瘦页 (点更新/小表) 则能 // 在一页里合并几十条, 页尾不再被零填充浪费。 // 读端本就按 EntryCount 逐条解析 (见 wal_recovery.go), 所以打包只改写端。 // ============================================================================= // DataPage 磁盘布局见 page.DataPageHdr (头 40B), 行数据区从 page.DataPageHdrSize 开始。 // WAL 状态常量 (对应 State 字段) const ( WALStateMem = 0 // 内存中,未写盘 WALStateWritten = 1 // 已 write(),未 fsync() WALStatePersisted = 2 // 已 fsync(),但未标记提交 WALStateCommitted = 3 // 已提交,可安全恢复 ) // ============================================================================= // WAL 条目操作类型 (每个条目都显式带 1B OpType) // // INSERT: Data = storeRow [pkID(8B BE) + rowData] // DELETE: Data = pkID(8B BE) // UPDATE: Data = storeRow [pkID(8B BE) + rowData] (新行数据) // ============================================================================= const ( WALOpFullPage = 0 // 整页镜像 (Data = 页内容) WALOpInsert = 1 // 插入行 WALOpDelete = 2 // 删除行 WALOpUpdate = 3 // 更新行 ) // RowEntry 行级 WAL 记录 type RowEntry struct { PageID uint32 // 目标 DataPage ID OpType byte // WALOpInsert / WALOpDelete / WALOpUpdate StoreRow []byte // INSERT/UPDATE: [pkID(8B BE) + rowData]; DELETE: pkID(8B BE) } // WALWriter 事务 WAL 写入器 type WALWriter struct { txnID uint64 entries []walEntry file *File header *HeaderPage state byte // 当前 WAL 状态 (Mem/Written/Persisted/Committed) } type walEntry struct { pageID uint32 data []byte // 完整页镜像 (isRowLevel=false 时使用) isRowLevel bool // 是否为行级记录(true=存行数据,false=存整页) opType byte // 行级操作类型 (WALOpInsert/WALOpDelete/WALOpUpdate) rowEntry *RowEntry // 行级记录 } // BeginTxn 开始新事务 func (db *DB) BeginTxn() *WALWriter { // NextTxnID 是头页里的全局游标, 提交 / 事务提交都可能并发取号, 必须在 headerMu 内自增。 var txnID uint64 db.file.withHeader(func() { txnID = db.header.NextTxnID db.header.NextTxnID++ }) return &WALWriter{ txnID: txnID, entries: make([]walEntry, 0, 16), file: db.file, header: db.header, state: WALStateMem, // 初始状态: Mem (内存中) } } // RecordPage 记录一个被修改的页 (同一 pageID 只保留最新镜像) func (w *WALWriter) RecordPage(pageID uint32, data []byte) { d := make([]byte, len(data)) copy(d, data) // 同一 pageID 只保留最新镜像 for i := range w.entries { if w.entries[i].pageID == pageID { w.entries[i].data = d return } } w.entries = append(w.entries, walEntry{ pageID: pageID, data: d, isRowLevel: false, // 原有逻辑:存整页 }) } // RecordRowChange 记录行级变更 (带 OpType, 用于崩溃恢复) // pageID: 目标 DataPage ID // opType: WALOpInsert / WALOpDelete / WALOpUpdate // storeRow: INSERT/UPDATE = [pkID(8B BE) + rowData], DELETE = pkID(8B BE) func (w *WALWriter) RecordRowChange(pageID uint32, opType byte, storeRow []byte) { data := make([]byte, len(storeRow)) copy(data, storeRow) w.entries = append(w.entries, walEntry{ pageID: pageID, isRowLevel: true, opType: opType, rowEntry: &RowEntry{ PageID: pageID, OpType: opType, StoreRow: data, }, }) } // Rollback 回滚: 丢弃所有条目 func (w *WALWriter) Rollback() { w.entries = nil w.state = WALStateMem // 重置状态 } // Commit 原子提交流程: // 1. 将 entries 写入 WAL 页链 (State: Mem → Written) // 2. fsync WAL 页 (State: Written → Persisted) // 3. 标记 WAL 首页 Committed = 1 并 fsync (State: Persisted → Committed) // 4. 回写实际页 (redo; 整页镜像, 重复执行幂等) // 5. 回收所有 WAL 页 + 清空 Header 链指针 + fsync // // ⭐ 关键: 「提交标记」必须先于「回写真实页」落盘。 // 若顺序反过来, 断电落在两者之间时真实页已被原地覆盖, 而提交标记仍是旧值, // 恢复会丢弃整个 WAL, 留下半应用(损坏)的页 —— 这里没有 undo, 无法自愈。 // // 缓冲池的脏标记【不在这里】收尾: 提交方拿到 success 之后调用 File.MarkClean // (按版本号逐页判定), 因为只有提交方知道哪些页属于这次快照, 也才知道窗口里有没有 // 人把这些页又改了一遍。 func (w *WALWriter) Commit() error { if len(w.entries) == 0 { return nil } // === 步骤 1: 写入 WAL 页链 (State: Mem → Written) === walPageIDs, err := w.writeWALPages() if err != nil { return err } // === 步骤 2: fsync WAL 页 (State: Written → Persisted) === if err := w.file.sync(); err != nil { return err } // 状态转换: Written → Persisted (已落盘) w.state = WALStatePersisted // === 步骤 3: 标记 WAL 首页为已提交, 并 fsync (State: Persisted → Committed) === // 这一步之后, 无论发生什么崩溃, 恢复都能重放整笔事务 (WAL 已完整且被标记)。 if err := w.markCommitted(walPageIDs[0]); err != nil { return err } if err := w.file.sync(); err != nil { return err } // === 步骤 4: 回写实际页 (直接写盘, 绕过 BufferPool) === // 放在提交标记之后: 中途断电可由恢复阶段重做 (整页镜像, 幂等)。 // // 用 WritePageHome 而不是 WritePageDirect: 这里写出去的正是缓存快照本身, 缓存里 // 那一份的内容只会等于它或新于它 (并发写入时), 无差别失效会丢掉更新的那一份。 // 缓存状态由提交方在 Commit() 返回后用 File.MarkClean 按版本号收尾 —— 这也是 // 这里不再 ClearCache 的原因: ClearAll 会把提交窗口里被重新改脏的页一并标干净, // 那些页并不在本笔 WAL 里, 标干净等于把它们的新内容永久丢掉。 for _, e := range w.entries { if e.pageID == 0 { continue } if err := w.file.WritePageHome(e.pageID, e.data); err != nil { return err } } // === 步骤 5: 回收整条 WAL 链 (O(1) 拼接, 与链长无关) === // 链内每页的 next 都写在 PageHdr.NextPage, 与空闲链同字段, 所以整条链 // head…tail 可以直接挂到空闲链尾, 不必逐页遍历回收。 // 链首/链尾/链长都已在 writeWALPages 中记录 (walPageIDs), 尾页 NextPage 必为 0。 // // 还链 (改空闲链指针) 与清空 WAL 链指针是一次头页上的复合修改, 必须整体落在 // headerMu 内, 否则并发写入线程可能看到"WAL 链已清空、空闲链却还没接好"的中间态。 var recycleErr error w.file.withHeader(func() { recycleErr = recycleChainWithCache(w.file, w.header, walPageIDs[0], walPageIDs[len(walPageIDs)-1], uint32(len(walPageIDs))) if recycleErr != nil { return } w.header.FirstWALPage = 0 w.header.LastWALPage = 0 recycleErr = w.file.writeHeaderLocked(w.header) }) if recycleErr != nil { return recycleErr } // 收尾必须 fsync: 否则断电后磁盘上的 Header 仍指向已提交的 WAL, // 下次打开会重复重放 (行级记录场景还会造成重复插入) if err := w.file.sync(); err != nil { return err } w.entries = nil return nil } // writeWALPages 将所有 entry 紧凑写入 WAL 页链, 返回分配的页 ID 列表 // 状态转换: Mem → Written // // 打包策略: 一条 Entry 占 [13B 元数据 + DataLen]。当前页剩余空间够就继续追加, // 不够才定型落盘并开新页 —— 一条 200B 的记录原本独占 16KB, 现在一页可放约 76 条。 // // 所有 WAL 页头部等长 (page.WALPageHdrSize); walEntryPayload 已把单条记录截断到 // 「一定能放进空页」的长度, 因此开新页后必定写得下, 不会出现空页仍放不下的情况。 func (w *WALWriter) writeWALPages() ([]uint32, error) { var pageIDs []uint32 // 当前正在填充的 WAL 页 var curBuf []byte var curPID uint32 var curIsFirst bool var curOff uint32 // 已用字节数 (从页头之后开始) var curEntries uint16 // 已写入的条目数 // flushCurrent 定型当前页 (回填 UseSize / EntryCount) 并落盘。 // 链指针不在这里填 —— 只有"开新页"时才知道下一页是谁, 由调用处先回填再 flush。 // ⭐ 全程用内存缓冲记录"是否首页", 不从磁盘读回上一页判定 —— 那种实现依赖 // TxnID != 0, 而首个事务的 TxnID 恰好是 0, 会把链指针写到错误偏移。 flushCurrent := func() error { if curBuf == nil { return nil } wh := page.WalHdrOf(curBuf) wh.UseSize = curOff wh.EntryCount = curEntries return w.file.WritePageDirect(curPID, curBuf) } for i := range w.entries { opType, targetPID, payload := w.walEntryPayload(w.entries[i]) dataLen := uint32(len(payload)) need := page.WALEntryMeta + dataLen // 当前页放不下 → 开新页 if curBuf == nil || uint32(PageSize)-curOff < need { // 分配新页: headerMu 内摘空闲链 / 递增游标并推进头页。 // erase=false —— 页紧接着就在内存里被整页重造 (curBuf 是全 0 的新切片, // flushCurrent 一次写满 16KB), 不需要擦页; 且此刻正处在 Commit 内 // (持有 commitMu), 擦页会重入 autoCommit → Commit → 自死锁在 commitMu 上。 pid := w.file.allocPage(w.header, false) // 先回填上一页的链指针, 再把上一页定型落盘。 // 链指针统一写 PageHdr.NextPage (偏移 12) —— 与空闲链表同一个字段, // 所以整条 WAL 链本身就是一条合法空闲链, 回收时可一次拼接 (见 Commit 步骤 5)。 if curBuf != nil { page.HdrOf(curBuf).NextPage = pid if err := flushCurrent(); err != nil { return nil, err } } pageIDs = append(pageIDs, pid) curBuf = make([]byte, PageSize) curPID = pid curIsFirst = len(pageIDs) == 1 curEntries = 0 var prevPID uint32 if len(pageIDs) > 1 { prevPID = pageIDs[len(pageIDs)-2] } wh := page.WalHdrOf(curBuf) wh.Init(pid, prevPID) if curIsFirst { // === 首页: TxnID + TotalEntries + State === wh.TxnID = w.txnID wh.TotalEntries = uint32(len(w.entries)) wh.State = WALStateWritten // State = Written (已写盘,未 fsync) } // 首页与续页头部等长, 条目区起点一致 curOff = page.WALPageHdrSize } // 追加当前 entry // 统一格式: [TargetPageID(4B)][DataLen(4B)][Checksum(4B)][OpType(1B)][Data] // CRC32 覆盖 OpType + Data, 整页/行级一致, 类型显式存储不再靠首字节猜测。 csBuf := make([]byte, 1+dataLen) csBuf[0] = opType copy(csBuf[1:], payload) binary.LittleEndian.PutUint32(curBuf[curOff:], targetPID) binary.LittleEndian.PutUint32(curBuf[curOff+4:], dataLen) binary.LittleEndian.PutUint32(curBuf[curOff+8:], crc32.ChecksumIEEE(csBuf)) curBuf[curOff+12] = opType curOff += page.WALEntryMeta copy(curBuf[curOff:], payload) curOff += dataLen curEntries++ } // 定型最后一页 if err := flushCurrent(); err != nil { return nil, err } // 更新 HeaderPage 的 WAL 链指针 // 发布链指针是为了崩溃恢复能找回这笔事务, 必须在 headerMu 内对"首页是否已存在" // 与"尾页指向"做一次原子更新, 否则并发提交可能互相覆盖链尾。 if len(pageIDs) > 0 { w.file.withHeader(func() { if w.header.FirstWALPage == 0 { w.header.FirstWALPage = pageIDs[0] } w.header.LastWALPage = pageIDs[len(pageIDs)-1] _ = w.file.writeHeaderLocked(w.header) }) } // 状态转换: Mem → Written (已写盘,未 fsync) w.state = WALStateWritten return pageIDs, nil } // walEntryPayload 解析一条 entry 的目标页 / OpType / 载荷, 并把载荷截断到 // 「一定能放进一张空 WAL 页」的长度 (首页/续页头等长, 只需一个限制)。 // // 这个截断是 writeWALPages 打包逻辑的前提: 开新页后必定能写下当前这条记录。 func (w *WALWriter) walEntryPayload(e walEntry) (byte, uint32, []byte) { opType := byte(WALOpFullPage) targetPID := e.pageID var payload []byte switch { case e.isRowLevel && e.rowEntry != nil: opType = e.rowEntry.OpType targetPID = e.rowEntry.PageID payload = e.rowEntry.StoreRow case e.isRowLevel: // 行级条目缺少 rowEntry: 视作空行级记录 opType = e.opType payload = nil default: // 整页记录: 只存有效前缀 (整页 16KB 无法与条目元数据同放一页) storeLen := uint32(len(e.data)) if e.pageID == 0 { // HeaderPage 只有前 4KB 有效; 必须与 recoverHeader 的解析长度一致 storeLen = PageSize4K } else if len(e.data) >= int(page.PageHdrSize) { useSize := page.HdrOf(e.data).UseSize if useSize > 0 && useSize < uint32(len(e.data)) { storeLen = useSize } } if storeLen > uint32(len(e.data)) { storeLen = uint32(len(e.data)) } payload = e.data[:storeLen] } // 防御: 单条记录不得超过空首页容量 (行级记录同样受此保护) if maxLen := uint32(PageSize) - page.WALPageHdrSize - page.WALEntryMeta; uint32(len(payload)) > maxLen { payload = payload[:maxLen] } return opType, targetPID, payload } // markCommitted 标记 WAL 首页为已提交 (State: Persisted → Committed) func (w *WALWriter) markCommitted(firstPageID uint32) error { buf := make([]byte, PageSize) // 必须直接读盘: WAL 页是用 WritePageDirect 写的, 走 BufferPool 可能读到陈旧缓存 if err := w.file.readPageDisk(firstPageID, buf); err != nil { return err } page.WalHdrOf(buf).State = WALStateCommitted // State = Committed if err := w.file.WritePageDirect(firstPageID, buf); err != nil { return err } // 状态转换: Persisted → Committed w.state = WALStateCommitted return nil } // sync 刷盘 (fsync) func (ff *File) sync() error { return ff.f.Sync() } // ============================================================================= // applyRowToPage 将行级 WAL 记录应用到页缓冲区 (用于崩溃恢复) // // 操作类型: // // WALOpInsert: 在 DataPage 中追加一行 // WALOpDelete: 在 DataPage 中软删除指定 pkID 的行 // WALOpUpdate: 在 DataPage 中按 pkID 原地更新行数据 // // storeRow 格式: [pkID(8B BE) + rowData] (INSERT/UPDATE) // storeRow 格式: pkID(8B BE) (DELETE) // ============================================================================= func applyRowToPage(buf []byte, opType byte, storeRow []byte) { const delFlag = uint32(0x80000000) if len(buf) < int(page.DataPageHdrSize) { return } // 读取 DataPage 元数据 (Count/Min/Max 紧随 PageHdr) dh := page.DataHdrOf(buf) count := dh.Count dataEnd := dh.UseSize if dataEnd < page.DataPageHdrSize || dataEnd > uint32(len(buf)) { return } switch opType { case WALOpInsert: // 追加行: [len(4B)] [storeRow] off := dataEnd if off+4+uint32(len(storeRow)) > uint32(len(buf)) { return // 页满, 放弃 (整页镜像已包含该行) } binary.LittleEndian.PutUint32(buf[off:], uint32(len(storeRow))) off += 4 copy(buf[off:], storeRow) off += uint32(len(storeRow)) // 更新 count 和 dataEnd (UseSize) dh.Count = count + 1 dh.UseSize = off case WALOpDelete: // storeRow = pkID(8B BE) if len(storeRow) < 8 { return } keyPK := binary.BigEndian.Uint64(storeRow[:8]) off := page.DataPageHdrSize for i := uint32(0); i < count; i++ { if off+4 > dataEnd { break } rawLen := binary.LittleEndian.Uint32(buf[off:]) deleted := rawLen&delFlag != 0 rowLen := rawLen & ^delFlag rowStart := off + 4 if rowStart+rowLen > dataEnd { break } if !deleted && rowLen >= 8 { rowPK := binary.BigEndian.Uint64(buf[rowStart : rowStart+8]) if rowPK == keyPK { // 设置软删除标志 binary.LittleEndian.PutUint32(buf[off:], rowLen|delFlag) return } } off = rowStart + rowLen } case WALOpUpdate: // storeRow = [pkID(8B BE) + rowData] if len(storeRow) < 8 { return } keyPK := binary.BigEndian.Uint64(storeRow[:8]) off := page.DataPageHdrSize for i := uint32(0); i < count; i++ { if off+4 > dataEnd { break } rawLen := binary.LittleEndian.Uint32(buf[off:]) deleted := rawLen&delFlag != 0 rowLen := rawLen & ^delFlag rowStart := off + 4 if rowStart+rowLen > dataEnd { break } if !deleted && rowLen >= 8 { rowPK := binary.BigEndian.Uint64(buf[rowStart : rowStart+8]) if rowPK == keyPK { // 原地覆盖行数据 copy(buf[rowStart:], storeRow) // 剩余字节置零 for j := uint32(len(storeRow)); j < rowLen; j++ { buf[rowStart+j] = 0 } return } } off = rowStart + rowLen } } }