| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627 |
- package core
- import (
- "git.ycl.pub/yl32940/TsDB2/DB/core/index/datapage"
- "git.ycl.pub/yl32940/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 == WALOpPageRest(4) → Data = [覆盖偏移 u32][页尾原文], 把截掉的
- // 页尾按偏移覆盖回去 —— 槽位目录/被容量上限
- // 截掉的尾数据由此恢复, 见 packEntries
- // 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 = 页内容有效前缀, 尾部由 WALOpPageRest 补齐)
- WALOpInsert = 1 // 插入行
- WALOpDelete = 2 // 删除行
- WALOpUpdate = 3 // 更新行
- // WALOpPageRest: 整页镜像的"补尾"记录。Data = [覆盖偏移 u32][页尾原文], 必须
- // 跟在同页 WALOpFullPage 之后 (恢复端用"同页已有 base"作存在性校验, 偏移本身
- // 显式存储在 Data 头部)。
- //
- // 为什么需要它: 整页镜像只存有效前缀 (useSize), 而 DataPage 的槽位目录在页尾
- // (useSize 之外) —— 重放时零填充把目录抹掉, 页上全部行经槽位不可达 (chaos 压测
- // 实锤: 崩溃恢复后 TotalRows 与实际可达行数分叉)。Heap/Blob 满页同理会被
- // "单条容量上限"截掉真实数据。补尾记录让截掉的每一字节都可恢复。
- WALOpPageRest = 4
- )
- // 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
- // pageIndex: pageID → entries 下标。RecordPage 的"同页只留最新镜像"去重用,
- // 免掉每次提交 O(条目数²) 的线性扫 (默认阈值 500 页/提交时是 12.5 万次比较)。
- pageIndex map[uint32]int
- 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),
- pageIndex: make(map[uint32]int, 16),
- file: db.file,
- header: db.header,
- state: WALStateMem, // 初始状态: Mem (内存中)
- }
- }
- // RecordPage 记录一个被修改的页 (同一 pageID 只保留最新镜像)。
- // 去重走 pageIndex 映射, O(1); 旧实现对每条 entry 线性扫全部 entries,
- // 一次 500 页的提交就是 12.5 万次比较。
- func (w *WALWriter) RecordPage(pageID uint32, data []byte) {
- d := make([]byte, len(data))
- copy(d, data)
- if i, ok := w.pageIndex[pageID]; ok {
- w.entries[i].data = d
- return
- }
- w.pageIndex[pageID] = len(w.entries)
- w.entries = append(w.entries, walEntry{
- pageID: pageID,
- data: d,
- isRowLevel: false, // 原有逻辑:存整页
- })
- }
- // Rollback 回滚: 丢弃所有条目
- func (w *WALWriter) Rollback() {
- w.entries = nil
- w.pageIndex = nil
- w.state = WALStateMem // 重置状态
- }
- // Commit 原子提交流程:
- // 1. 将 entries 写入 WAL 页链 (State: Mem → Written)
- // 2. 标记 WAL 首页 Committed = 1 (State → Committed)
- // 3. fsync 一次: WAL 页与提交标记随这次 fsync 一起落盘
- // (排序要求的是"标记 durable ⇒ 数据已 durable", 同一次 fsync 前的顺序写出即满足)
- // 4. 回写实际页 (redo; 整页镜像, 重复执行幂等)
- // 5. 回收所有 WAL 页 + 清空 Header 链指针 + fsync
- //
- // ⭐ 关键: 「提交标记」必须先于「回写真实页」落盘。
- // 若顺序反过来, 断电落在两者之间时真实页已被原地覆盖, 而提交标记仍是旧值,
- // 恢复会丢弃整个 WAL, 留下半应用(损坏)的页 —— 这里没有 undo, 无法自愈。
- //
- // 缓冲池的脏标记【不在这里】收尾: 提交方拿到 success 之后调用 File.MarkClean
- // (按版本号逐页判定), 因为只有提交方知道哪些页属于这次快照, 也才知道窗口里有没有
- // 人把这些页又改了一遍。
- func (w *WALWriter) Commit() error {
- if len(w.entries) == 0 {
- return nil
- }
- // === 步骤 1+2+3: 写 WAL 页链 → 标记首页已提交 → 一次 fsync ===
- // WAL 数据与提交标记在同一次 fsync 前依次写出: fsync 保证"标记 durable ⇒
- // 数据先于它 durable"的顺序不变 (排序要求的是先后持久, 不是两次 fsync)。
- // 旧实现这里分两次 fsync (写完数据一次、标记后再一次), 每次提交白付一轮 fsync 延迟。
- walPageIDs, firstBuf, err := w.writeWALPages()
- if err != nil {
- return err
- }
- // === 步骤 3: 标记 WAL 首页为已提交 (State: Written → Committed) ===
- // 这一步之后, 无论发生什么崩溃, 恢复都能重放整笔事务 (WAL 已完整且被标记)。
- if err := w.markCommitted(walPageIDs[0], firstBuf); err != nil {
- return err
- }
- // 状态转换: Written → Persisted → Committed (随下面这次 fsync 落盘)
- if err := w.file.sync(); err != nil {
- return err
- }
- w.state = WALStateCommitted
- // === 步骤 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
- }
- }
- // 步骤 4.5: 回写后立刻 fsync, 钉死"数据页先于头页(步骤 5 清 WAL 指针)落盘"。
- // 没有这一次 fsync, OS 后台回刷可能把步骤 5 的头页先刷下去、数据页留在缓存 —
- // 断电后头页说"没有 WAL 可重放", 数据页却是旧值, 已提交数据静默丢失。
- if err := w.file.sync(); 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, 它钉死一个关键的持久化顺序: "忘掉日志"(头页: WAL 链回收
- // + 指针清零) 不得先于"数据就位"(步骤 4 回写的真实数据页) 落盘。
- // 没有这次 fsync, 断电时 OS 可能只把头页刷下去、数据页留在缓存里 —— 重开时
- // 头页说"没有 WAL", 已提交的数据却是旧内容, 且 WAL 页已被标成空闲页即将被
- // 覆盖, 连重放救回的机会都没有 (静默丢已提交数据)。
- // 顺带的效果: 头页若仍指着已提交的 WAL, 下次打开会重复重放 —— 整页镜像幂等
- // 无害, 行级记录则会重复插入。
- if err := w.file.sync(); err != nil {
- return err
- }
- w.entries = nil
- w.pageIndex = nil
- return nil
- }
- // writeWALPages 将所有 entry 紧凑写入 WAL 页链, 返回 (分配的页 ID 列表, 首页缓冲区)。
- // 状态转换: Mem → Written
- //
- // 打包策略: 一条 Entry 占 [13B 元数据 + DataLen]。当前页剩余空间够就继续追加,
- // 不够才定型落盘并开新页 —— 一条 200B 的记录原本独占 16KB, 现在一页可放约 76 条。
- //
- // 所有 WAL 页头部等长 (page.WALPageHdrSize); walEntryPayload 已把单条记录截断到
- // 「一定能放进空页」的长度, 因此开新页后必定写得下, 不会出现空页仍放不下的情况。
- func (w *WALWriter) writeWALPages() ([]uint32, []byte, error) {
- var pageIDs []uint32
- var firstBuf []byte // 首页缓冲区: markCommitted 直接改它打提交标记, 免一次整页读回
- // 当前正在填充的 WAL 页
- var curBuf []byte
- var curPID uint32
- var curIsFirst bool
- var curOff uint32 // 已用字节数 (从页头之后开始)
- var curEntries uint16 // 已写入的条目数
- var opTypeBuf [1]byte // CRC 的 opType 段缓冲 (分段 Update 免每条一次临时分配)
- // 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)
- }
- packed := w.packEntries()
- for i := range packed {
- opType, targetPID, payload := packed[i].opType, packed[i].pageID, packed[i].payload
- 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, nil, err
- }
- }
- pageIDs = append(pageIDs, pid)
- curBuf = make([]byte, PageSize)
- if len(pageIDs) == 1 {
- // 首页缓冲留一份给 markCommitted: 它的内容在"开第二页回填链指针 +
- // flush"后就定稿了, 之后不再被本函数改动 —— 提交标记直接改这份内存
- // 缓冲写回, 免一次"整页读回只为改 1 字节 State"。
- firstBuf = curBuf
- }
- 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(packed)) // 展开后 (含 PageRest) 的记录总数
- wh.State = WALStateWritten // State = Written (已写盘,未 fsync)
- }
- // 首页与续页头部等长, 条目区起点一致
- curOff = page.WALPageHdrSize
- }
- // 追加当前 entry
- // 统一格式: [TargetPageID(4B)][DataLen(4B)][Checksum(4B)][OpType(1B)][Data]
- // CRC32 覆盖 OpType + Data, 整页/行级一致, 类型显式存储不再靠首字节猜测。
- // crc32.Update 分段计算与"拼成一条再 ChecksumIEEE"完全等价, 省掉每条一次
- // 1+DataLen 的临时缓冲分配。
- opTypeBuf[0] = opType
- crc := crc32.Update(0, crc32.IEEETable, opTypeBuf[:])
- crc = crc32.Update(crc, crc32.IEEETable, payload)
- binary.LittleEndian.PutUint32(curBuf[curOff:], targetPID)
- binary.LittleEndian.PutUint32(curBuf[curOff+4:], dataLen)
- binary.LittleEndian.PutUint32(curBuf[curOff+8:], crc)
- curBuf[curOff+12] = opType
- curOff += page.WALEntryMeta
- copy(curBuf[curOff:], payload)
- curOff += dataLen
- curEntries++
- }
- // 定型最后一页
- if err := flushCurrent(); err != nil {
- return nil, nil, err
- }
- // 更新 HeaderPage 的 WAL 链指针
- // 发布链指针是为了崩溃恢复能找回这笔事务, 必须在 headerMu 内对"首页是否已存在"
- // 与"尾页指向"做一次原子更新, 否则并发提交可能互相覆盖链尾。
- if len(pageIDs) > 0 {
- var hdrErr error
- w.file.withHeader(func() {
- if w.header.FirstWALPage == 0 {
- w.header.FirstWALPage = pageIDs[0]
- }
- w.header.LastWALPage = pageIDs[len(pageIDs)-1]
- hdrErr = w.file.writeHeaderLocked(w.header)
- })
- if hdrErr != nil {
- // 链指针发布失败: 这批 WAL 页不会被恢复找到, 必须让提交失败,
- // 不能让脏页带着"盘上无任何记录"的状态继续跑 (静默丢数据)
- return nil, nil, hdrErr
- }
- }
- // 状态转换: Mem → Written (已写盘,未 fsync)
- w.state = WALStateWritten
- return pageIDs, firstBuf, nil
- }
- // packedWalEntry 打包后的最终 WAL 记录 (writeWALPages 的输入)。
- // 每条 = [13B 元数据 + payload], 与 w.entries 的区别: 整页镜像可能被展开成
- // FullPage + PageRest 两条 (见 packEntries)。
- type packedWalEntry struct {
- opType byte
- pageID uint32
- payload []byte
- }
- // packEntries 把逻辑 entries 展开成最终记录序列。
- //
- // 整页镜像在「useSize 有效前缀截断」或「单条容量上限 (PageSize-40-13)」时会丢掉
- // 页尾字节, 而其中可能装着 recoverable 的数据:
- //
- // - DataPage 的槽位目录 (页尾, useSize 之外): 被抹掉后页上全部行不可达;
- // - Heap/Blob 满页被容量上限截掉的尾数据: 直接丢字节。
- //
- // 对这些页补一条 WALOpPageRest, 把 [max(baseLen, 目录起始), 页尾) 的原文带上。
- // 恢复端按"同页上一条 FullPage 的 payload 长度"定位覆盖偏移, 因此补尾必须紧跟
- // 在对应整页镜像之后 (此处按序展开天然保证)。
- func (w *WALWriter) packEntries() []packedWalEntry {
- var out []packedWalEntry
- for i := range w.entries {
- e := &w.entries[i]
- opType, targetPID, payload := w.walEntryPayload(*e)
- out = append(out, packedWalEntry{opType: opType, pageID: targetPID, payload: payload})
- if e.isRowLevel || targetPID == 0 {
- continue // 行级记录与头页镜像 (4KB 完整) 没有截断问题
- }
- pageLen := uint32(len(e.data))
- if uint32(len(payload)) >= pageLen {
- continue // 整页入链, 无截断
- }
- var restStart uint32
- hdr := page.HdrOf(e.data)
- if hdr.UseSize > uint32(len(payload)) {
- // 容量上限截掉了 [len(payload), useSize) 的真实数据 (行/堆块/blob 字节),
- // 必须补到页尾 —— DataPage 的目录也在其中, 一并覆盖。
- restStart = uint32(len(payload))
- } else if hdr.Type == page.DataIndex {
- if ds, ok := datapage.SlotDirStart(e.data); ok && ds > uint32(len(payload)) {
- // useSize 处截断: [useSize, 目录起始) 是未使用的零区, 重放零填充无害;
- // 但页尾的槽位目录必须随链可恢复, 否则页上全部行经槽位不可达。
- restStart = ds
- } else {
- continue // 目录已在 base 内或无效 (Load 按无槽位处理), 无需补尾
- }
- } else {
- continue // useSize 之内完整入链, 尾部未使用
- }
- if restStart < pageLen {
- // Data = [覆盖偏移 u32][页尾原文]: 偏移显式入链, 恢复端不依赖"紧跟同页
- // base 之后"来推导位置 (截断点与目录起始经常不一致)
- rest := make([]byte, 4+pageLen-restStart)
- binary.LittleEndian.PutUint32(rest, restStart)
- copy(rest[4:], e.data[restStart:pageLen])
- out = append(out, packedWalEntry{
- opType: WALOpPageRest,
- pageID: targetPID,
- payload: rest,
- })
- }
- }
- return out
- }
- // walEntryPayload 解析一条 entry 的目标页 / OpType / 载荷, 并把载荷截断到
- // 「一定能放进一张空 WAL 页」的长度 (首页/续页头等长, 只需一个限制)。
- //
- // 这个截断是 writeWALPages 打包逻辑的前提: 开新页后必定能写下当前这条记录。
- // 注意: 被截掉的页尾不丢失 —— packEntries 会为其补一条 WALOpPageRest。
- 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 → Committed)。
- // firstBuf 是 writeWALPages 手里刚写盘的首页内存缓冲 (内容与盘上逐字节一致),
- // 直接改它的 State 字节整页写回 —— 旧实现为此整页读回 16KB, 只为改 1 字节。
- func (w *WALWriter) markCommitted(firstPageID uint32, firstBuf []byte) error {
- if len(firstBuf) != int(PageSize) {
- // 防御: 缓冲缺失/长度不符时退回读盘路径 (正常不会发生)
- firstBuf = make([]byte, PageSize)
- // 必须直接读盘: WAL 页是用 WritePageDirect 写的, 走 BufferPool 可能读到陈旧缓存
- if err := w.file.readPageDisk(firstPageID, firstBuf); err != nil {
- return err
- }
- }
- page.WalHdrOf(firstBuf).State = WALStateCommitted // State = Committed
- if err := w.file.WritePageDirect(firstPageID, firstBuf); err != nil {
- return err
- }
- 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); 追加的是活行, 存活计数同步 +1
- dh.Count = count + 1
- dh.LiveCnt++
- 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 {
- // 设置软删除标志; 存活/失效统计同步维护 (页头持久化, 见 page.DataPageHdr)
- binary.LittleEndian.PutUint32(buf[off:], rowLen|delFlag)
- dh.LiveCnt--
- dh.Dead += 4 + rowLen
- 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
- }
- }
- }
|