package core import ( "fmt" ) // Transaction 事务管理器 // // 核心思想: // - 事务开始时,记录所有被修改行的原始数据和新数据 // - 事务内操作正常修改 BufferPool // - Commit 时将行级数据写入 WAL(节省空间) // - Rollback 时用原始数据覆盖 BufferPool 中的脏页(立即生效) type Transaction struct { db *DB originalPages map[uint32][]byte // pageID → 原始页数据快照(用于 Rollback) rowChanges map[uint32]struct{ original, new []byte } // pageID → {原始行, 新行}(用于 WAL) committed bool rolledBack bool } // BeginTransaction 开启新事务 func (db *DB) BeginTransaction() *Transaction { txn := &Transaction{ db: db, originalPages: make(map[uint32][]byte), rowChanges: make(map[uint32]struct{ original, new []byte }), committed: false, rolledBack: false, } // 设置当前活跃事务 db.currentTxn = txn return txn } // cacheOriginalPage 缓存页的原始数据(首次修改时调用) func (txn *Transaction) cacheOriginalPage(pageID uint32) error { // 如果已经缓存过,不需要重复缓存 if _, exists := txn.originalPages[pageID]; exists { return nil } // 从磁盘读取原始数据 buf := make([]byte, PageSize) if err := txn.db.file.readPageDisk(pageID, buf); err != nil { return fmt.Errorf("failed to read page %d for rollback cache: %v", pageID, err) } txn.originalPages[pageID] = buf return nil } // RecordRowChange 记录行级变化(用于 WAL) // originalRow: 原始行数据 // newRow: 新行数据 func (txn *Transaction) RecordRowChange(pageID uint32, originalRow []byte, newRow []byte) { txn.rowChanges[pageID] = struct{ original, new []byte }{ original: originalRow, new: newRow, } } // Commit 提交事务 // 1. 收集所有 BufferPool 脏页 // 2. 写入 WAL // 3. ⭐ 关键:fsync 确保落盘 ⭐ // 4. 回收 WAL 页 // 5. 清理原始数据快照(释放内存) func (txn *Transaction) Commit() error { if txn.committed { return fmt.Errorf("transaction already committed") } if txn.rolledBack { return fmt.Errorf("transaction already rolled back") } // 1. 创建 WAL 事务 wal := txn.db.BeginTxn() // 2. 记录行级数据到 WAL(如果有的话) if len(txn.rowChanges) > 0 { for pageID, change := range txn.rowChanges { wal.RecordRowLevel(pageID, change.original, change.new) } } else { // Fallback: 整页模式 dirty := txn.db.file.CollectDirty() if len(dirty) == 0 { txn.committed = true txn.db.currentTxn = nil txn.originalPages = nil txn.rowChanges = nil return nil // 无脏页,直接返回 } for _, d := range dirty { wal.RecordPage(d.PageID, d.Data) } } // 4. ⭐ 关键:Commit 会触发 fsync,确保落盘 ⭐ if err := wal.Commit(); err != nil { return fmt.Errorf("WAL commit failed: %v", err) } // 5. 清理事务状态 + 释放原始数据快照 txn.committed = true txn.db.currentTxn = nil txn.originalPages = nil // ← 提交成功,原始数据不再需要 txn.rowChanges = nil // ← 清理行级变化 return nil } // Rollback 回滚事务 // 用原始数据快照覆盖 BufferPool 中的脏页(立即生效) // 回滚后清理所有缓存数据(释放内存) func (txn *Transaction) Rollback() error { if txn.committed { return fmt.Errorf("transaction already committed, cannot rollback") } if txn.rolledBack { return fmt.Errorf("transaction already rolled back") } // 用原始数据覆盖所有被修改的页 for pageID, originalData := range txn.originalPages { // 直接写回 BufferPool(绕过 WAL) if err := txn.db.file.WritePage(pageID, originalData); err != nil { return fmt.Errorf("failed to rollback page %d: %v", pageID, err) } } // 清理事务状态 + 释放所有缓存数据 txn.rolledBack = true txn.db.currentTxn = nil txn.originalPages = nil // ← 回滚成功,原始数据和修改数据都不再需要 txn.rowChanges = nil // ← 清理行级变化 return nil } // IsCommitted 检查事务是否已提交 func (txn *Transaction) IsCommitted() bool { return txn.committed } // IsRolledBack 检查事务是否已回滚 func (txn *Transaction) IsRolledBack() bool { return txn.rolledBack }