package core import ( "TsDB2/DB/core/page" "encoding/binary" "hash/crc32" "unsafe" ) // ============================================================================= // WAL (Write-Ahead Log) 实现 // // WAL 页布局 (16KB): // 首页: [PageHdr(20B)] [TxnID(8B)] [TotalEntries(4B)] [State(1B)] // [NextWalPage(4B)] [EntryCount(2B)] [WALEntry...] // 续页: [PageHdr(20B)] [NextWalPage(4B)] [EntryCount(2B)] [WALEntry...] // // WALEntry: [TargetPageID(4B)] [DataLen(4B)] [Checksum(4B)] [Data(DataLen)] // // WAL 状态机 (State 字段): // 0 = Mem (内存中,未写盘) // 1 = Written (已 write(),未 fsync()) // 2 = Persisted (已 fsync(),但未标记提交) // 3 = Committed (已提交,可安全恢复) // // 每个 WAL 页存 1 条完整 16KB 页镜像 (因为 entry 12B + 16KB > 16KB - 39B) // ============================================================================= const ( walPHDR = uint32(unsafe.Sizeof(page.PageHdr{})) // 20 walFirstHdr = walPHDR + 8 + 4 + 1 + 4 + 2 // 39: PageHdr + TxnID + TotalEntries + State + NextWalPage + EntryCount walContHdr = walPHDR + 4 + 2 // 26: PageHdr + NextWalPage + EntryCount walEntryMeta = 4 + 4 + 4 // 12: TargetPageID + DataLen + Checksum ) // WAL 状态常量 (对应 State 字段) const ( WALStateMem = 0 // 内存中,未写盘 WALStateWritten = 1 // 已 write(),未 fsync() WALStatePersisted = 2 // 已 fsync(),但未标记提交 WALStateCommitted = 3 // 已提交,可安全恢复 ) // ============================================================================= // 行级 WAL 操作类型 // // 行级记录新格式 (在 walEntryMeta 之后加 1B OpType): // [TargetPageID(4B)] [DataLen(4B)] [Checksum(4B)] [OpType(1B)] [Data...] // // INSERT: Data = storeRow [pkID(8B BE) + rowData] // DELETE: Data = pkID(8B BE) // UPDATE: Data = storeRow [pkID(8B BE) + rowData] (新行数据) // ============================================================================= const ( 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 readPage func(uint32, []byte) error // 用于分配页时读取空闲链表 state byte // 当前 WAL 状态 (Mem/Written/Persisted/Committed) } type walEntry struct { pageID uint32 data []byte // 完整页镜像(原有逻辑,用于 Commit) originalData []byte // 原始行数据(事务回滚用)← 新增 newData []byte // 新行数据(事务提交用)← 新增 isRowLevel bool // 是否为行级记录(true=存数据,false=存整页)← 新增 opType byte // 行级操作类型 (WALOpInsert/WALOpDelete/WALOpUpdate) rowEntry *RowEntry // 行级记录 (新格式, 用于恢复) } // BeginTxn 开始新事务 func (db *DB) BeginTxn() *WALWriter { txnID := db.header.NextTxnID db.header.NextTxnID++ return &WALWriter{ txnID: txnID, entries: make([]walEntry, 0, 16), file: db.file, header: db.header, readPage: db.file.readPageDisk, // 必须直接读盘! 走 BufferPool 会被旧缓存干扰空闲链表 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, // 原有逻辑:存整页 }) } // RecordRowLevel 记录行级数据(用于事务支持) // 存储原始数据和新数据,而非整页,节省空间 func (w *WALWriter) RecordRowLevel(pageID uint32, originalData []byte, newData []byte) { orig := make([]byte, len(originalData)) copy(orig, originalData) newD := make([]byte, len(newData)) copy(newD, newData) // 同一 pageID 只保留最新 for i := range w.entries { if w.entries[i].pageID == pageID && w.entries[i].isRowLevel { w.entries[i].originalData = orig w.entries[i].newData = newD return } } w.entries = append(w.entries, walEntry{ pageID: pageID, originalData: orig, newData: newD, isRowLevel: true, // 行级记录 }) } // 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 拷贝到真实位置) // 4. 标记 WAL 首页 Committed = 1 (State: Persisted → Committed) // 5. 回收所有 WAL 页 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: 回写实际页 (直接写盘, 绕过 BufferPool) === for _, e := range w.entries { if e.pageID == 0 { continue } if err := w.file.WritePageDirect(e.pageID, e.data); err != nil { return err } } w.file.ClearCache() // === 步骤 4: 标记 WAL 首页为已提交 (State: Persisted → Committed) === if err := w.markCommitted(walPageIDs[0]); err != nil { return err } if err := w.file.sync(); err != nil { return err } // === 步骤 5: 回收所有 WAL 页 (直接写盘 + 同步 BufferPool 缓存) === for _, pid := range walPageIDs { w.header.recyclePage(pid, w.file.WritePageDirect) // 同步 BufferPool 缓存: recyclePage 通过 WritePageDirect 写盘, 绕过了 BufferPool // 必须从磁盘读回并更新缓存, 否则 newPageWithReader 从缓存读到旧数据导致页面无法回收 syncBuf := make([]byte, PageSize) _ = w.file.readPageDisk(pid, syncBuf) w.file.SyncCache(pid, syncBuf) } w.header.FirstWALPage = 0 w.header.LastWALPage = 0 if err := w.file.WriteHeader(w.header); err != nil { return err } w.entries = nil return nil } // writeWALPages 将所有 entry 写入 WAL 页链, 返回分配的页 ID 列表 // 状态转换: Mem → Written func (w *WALWriter) writeWALPages() ([]uint32, error) { var pageIDs []uint32 entryIdx := 0 for entryIdx < len(w.entries) { // 分配新 WAL 页 (必须用 newPageWithReader 推进空闲链表) pid := w.header.newPageWithReader(w.readPage) pageIDs = append(pageIDs, pid) buf := make([]byte, PageSize) hdr := (*page.PageHdr)(unsafe.Pointer(&buf[0])) hdr.Type = page.WALIndex hdr.PageID = pid if len(pageIDs) > 1 { hdr.PrePage = pageIDs[len(pageIDs)-2] } off := uint32(0) if entryIdx == 0 { // === 首页: 写入 TxnID + TotalEntries + State + NextWalPage + EntryCount === binary.LittleEndian.PutUint64(buf[walPHDR:], w.txnID) binary.LittleEndian.PutUint32(buf[walPHDR+8:], uint32(len(w.entries))) buf[walPHDR+12] = WALStateWritten // State = Written (已写盘,未 fsync) binary.LittleEndian.PutUint32(buf[walPHDR+13:], 0) // NextWalPage (稍后填) binary.LittleEndian.PutUint16(buf[walPHDR+17:], 0) // EntryCount (稍后填) off = walFirstHdr } else { // === 续页: 写入 NextWalPage + EntryCount === binary.LittleEndian.PutUint32(buf[walPHDR:], 0) // NextWalPage (稍后填) binary.LittleEndian.PutUint16(buf[walPHDR+4:], 0) // EntryCount (稍后填) off = walContHdr } // 写入当前 entry e := w.entries[entryIdx] if e.isRowLevel && e.rowEntry != nil { // === 新行级格式: [PageID][DataLen][Checksum][OpType(1B)][Data] === dataLen := uint32(len(e.rowEntry.StoreRow)) // Checksum 覆盖 OpType + Data csBuf := make([]byte, 1+dataLen) csBuf[0] = e.rowEntry.OpType copy(csBuf[1:], e.rowEntry.StoreRow) checksum := crc32.ChecksumIEEE(csBuf) binary.LittleEndian.PutUint32(buf[off:], e.rowEntry.PageID) binary.LittleEndian.PutUint32(buf[off+4:], dataLen) binary.LittleEndian.PutUint32(buf[off+8:], checksum) buf[off+12] = e.rowEntry.OpType off += 13 // PageID(4) + DataLen(4) + Checksum(4) + OpType(1) copy(buf[off:], e.rowEntry.StoreRow) off += dataLen } else if e.isRowLevel { // === 旧行级格式: [PageID][OrigLen][NewLen][Checksum][OriginalData][NewData] === origLen := uint32(len(e.originalData)) newLen := uint32(len(e.newData)) totalLen := origLen + newLen combined := make([]byte, 0, totalLen) combined = append(combined, e.originalData...) combined = append(combined, e.newData...) checksum := crc32.ChecksumIEEE(combined) binary.LittleEndian.PutUint32(buf[off:], e.pageID) binary.LittleEndian.PutUint32(buf[off+4:], origLen) binary.LittleEndian.PutUint32(buf[off+8:], newLen) binary.LittleEndian.PutUint32(buf[off+12:], checksum) off += 16 copy(buf[off:], e.originalData) off += origLen copy(buf[off:], e.newData) off += newLen } else { // === 整页记录: [PageID][DataLen][Checksum][Data] (原有逻辑) === storeLen := uint32(len(e.data)) // 默认存全部 if len(e.data) >= 20 { useSize := binary.LittleEndian.Uint32(e.data[16:20]) if useSize > 0 && useSize < uint32(len(e.data)) { storeLen = useSize } // HeaderPage (pageID=0): 至少存 PageSize4K if e.pageID == 0 && storeLen < PageSize4K { storeLen = PageSize4K } } checksum := crc32.ChecksumIEEE(e.data[:storeLen]) binary.LittleEndian.PutUint32(buf[off:], e.pageID) binary.LittleEndian.PutUint32(buf[off+4:], storeLen) binary.LittleEndian.PutUint32(buf[off+8:], checksum) copy(buf[off+12:], e.data[:storeLen]) // storeLen 可能 < len(e.data), 剩余字节保持零填充 off += walEntryMeta + storeLen } hdr.UseSize = off entryIdx++ // 立即设置当前页的 EntryCount = 1 (不依赖后续链接代码) if entryIdx == 1 { // 首页: EntryCount 在 walPHDR+17 binary.LittleEndian.PutUint16(buf[walPHDR+17:], 1) } else { // 续页: EntryCount 在 walPHDR+4 binary.LittleEndian.PutUint16(buf[walPHDR+4:], 1) } // 链接上一页的 NextWalPage if len(pageIDs) > 1 { prevPID := pageIDs[len(pageIDs)-2] prevBuf := make([]byte, PageSize) if err := w.file.ReadPage(prevPID, prevBuf); err == nil { if isFirstWALPage(prevBuf) { binary.LittleEndian.PutUint32(prevBuf[walPHDR+13:], pid) binary.LittleEndian.PutUint16(prevBuf[walPHDR+17:], 1) } else { binary.LittleEndian.PutUint32(prevBuf[walPHDR:], pid) binary.LittleEndian.PutUint16(prevBuf[walPHDR+4:], 1) } _ = w.file.WritePageDirect(prevPID, prevBuf) } } // 写入当前页 if err := w.file.WritePageDirect(pid, buf); err != nil { return nil, err } } // 更新 HeaderPage 的 WAL 链指针 if len(pageIDs) > 0 { if w.header.FirstWALPage == 0 { w.header.FirstWALPage = pageIDs[0] } w.header.LastWALPage = pageIDs[len(pageIDs)-1] _ = w.file.WriteHeader(w.header) } // 状态转换: Mem → Written (已写盘,未 fsync) w.state = WALStateWritten return pageIDs, nil } // markCommitted 标记 WAL 首页为已提交 (State: Persisted → Committed) func (w *WALWriter) markCommitted(firstPageID uint32) error { buf := make([]byte, PageSize) if err := w.file.ReadPage(firstPageID, buf); err != nil { return err } buf[walPHDR+12] = WALStateCommitted // State = Committed if err := w.file.WritePageDirect(firstPageID, buf); err != nil { return err } // 状态转换: Persisted → Committed w.state = WALStateCommitted return nil } // isFirstWALPage 判断 WAL 页是否为事务首页 func isFirstWALPage(buf []byte) bool { // 首页在 walPHDR 处有 TxnID (非零), 续页没有 txnID := binary.LittleEndian.Uint64(buf[walPHDR:]) return txnID != 0 } // 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 dpHdrSize = uint32(unsafe.Sizeof(page.PageHdr{})) + 4 + 8 + 8 // 38B const delFlag = uint32(0x80000000) // 读取 DataPage 元数据 count := *(*uint32)(unsafe.Pointer(&buf[unsafe.Sizeof(page.PageHdr{})])) dataEnd := *(*uint32)(unsafe.Pointer(&buf[0])) // PageHdr.UseSize switch opType { case WALOpInsert: // 追加行: [len(4B)] [storeRow] off := dataEnd binary.LittleEndian.PutUint32(buf[off:], uint32(len(storeRow))) off += 4 copy(buf[off:], storeRow) off += uint32(len(storeRow)) // 更新 count 和 dataEnd (UseSize) *(*uint32)(unsafe.Pointer(&buf[unsafe.Sizeof(page.PageHdr{})])) = count + 1 *(*uint32)(unsafe.Pointer(&buf[0])) = off case WALOpDelete: // storeRow = pkID(8B BE) if len(storeRow) < 8 { return } keyPK := binary.BigEndian.Uint64(storeRow[:8]) off := dpHdrSize 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 := dpHdrSize 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 } } }