Bladeren bron

feat: 实现完整事务支持(Rollback立即生效)

核心改进:
- BufferPool 添加 onPageWrite 回调机制
- Transaction 缓存原始页数据快照
- Rollback 时用原始数据覆盖脏页(立即生效)

实现流程:
1. BeginTransaction() → 创建 originalPages map
2. Insert/Update/Delete → WritePage 触发回调
3. 回调检查 currentTxn → cacheOriginalPage(pageID)
4. Commit → WAL + fsync 落盘
5. Rollback → 用 originalPages 覆盖 BufferPool

关键文件:
- DB/core/transaction.go: 事务管理器 + cacheOriginalPage
- DB/core/pool/buffer_pool.go: onPageWrite 回调
- DB/core/File.go: SetOnPageWriteCallback
- DB/core/db.go: 初始化时设置回调

测试结果:
✅ Commit: 5行插入并提交,数据持久化
✅ Rollback: 5行插入后回滚,立即清除
✅ 重开验证: 回滚的5行不存在(从10行→5行)

优势:
- Rollback 立即生效(无需重启)
- 实现简单(回调机制)
- 复用现有逻辑(BufferPool/WAL)
yl32940 1 maand geleden
bovenliggende
commit
5e1ec587e2
5 gewijzigde bestanden met toevoegingen van 73 en 33 verwijderingen
  1. 7 0
      DB/core/File.go
  2. 9 0
      DB/core/db.go
  3. 15 0
      DB/core/pool/buffer_pool.go
  4. 42 33
      DB/core/transaction.go
  5. BIN
      txn_test.db

+ 7 - 0
DB/core/File.go

@@ -40,6 +40,13 @@ func (ff *File) SetDirtyThreshold(threshold int, onThreshold func()) {
 	}
 }
 
+// SetOnPageWriteCallback 设置页写入回调(用于事务缓存原始数据)
+func (ff *File) SetOnPageWriteCallback(callback func(pageID uint32)) {
+	if ff.bp != nil {
+		ff.bp.SetOnPageWriteCallback(callback)
+	}
+}
+
 // LockWrites 锁定写入(禁止 Insert/Update/Delete)
 func (ff *File) LockWrites() {
 	if ff.bp != nil {

+ 9 - 0
DB/core/db.go

@@ -122,6 +122,15 @@ func Open(path string) (*DB, error) {
 		autoCommitDelay: DefaultAutoCommitDelay,
 	}
 	db.startAutoCommit()
+	
+	// 设置事务回调(在 BufferPool.WritePage 时缓存原始数据)
+	ff.SetOnPageWriteCallback(func(pageID uint32) {
+		// 如果有活跃事务,缓存原始页数据
+		if db.currentTxn != nil {
+			_ = db.currentTxn.cacheOriginalPage(pageID)
+		}
+	})
+	
 	// 脏页阈值: 超过时自动触发 Commit 刷盘, 防止内存暴涨
 	ff.SetDirtyThreshold(DefaultDirtyThreshold, func() {
 		_ = db.Commit()

+ 15 - 0
DB/core/pool/buffer_pool.go

@@ -39,6 +39,9 @@ type BufferPool struct {
 
 	// BufferPool 容量控制
 	capacityWait chan struct{} // 容量满时等待的 channel
+	
+	// 事务支持
+	onPageWrite func(pageID uint32) // 页写入时的回调(用于事务缓存原始数据)
 }
 
 // NewBufferPool 创建页面缓冲池
@@ -183,6 +186,11 @@ func (bp *BufferPool) ReadPage(pageID uint32, buf []byte) error {
 func (bp *BufferPool) WritePage(pageID uint32, buf []byte) error {
 	// 1. 等待写锁释放(Commit/Flush 期间阻塞)
 	bp.waitForWriteLock()
+	
+	// 事务支持:通知 Transaction 缓存原始数据
+	if bp.onPageWrite != nil {
+		bp.onPageWrite(pageID)
+	}
 
 	cached := make([]byte, len(buf))
 	copy(cached, buf)
@@ -253,3 +261,10 @@ func (bp *BufferPool) SetCapacity(capacity int) {
 	bp.capacity = capacity
 	bp.mu.Unlock()
 }
+
+// SetOnPageWriteCallback 设置页写入回调(用于事务缓存原始数据)
+func (bp *BufferPool) SetOnPageWriteCallback(callback func(pageID uint32)) {
+	bp.mu.Lock()
+	bp.onPageWrite = callback
+	bp.mu.Unlock()
+}

+ 42 - 33
DB/core/transaction.go

@@ -4,28 +4,27 @@ import (
 	"fmt"
 )
 
-// Transaction 事务管理器(极简版)
+// Transaction 事务管理器
 // 
 // 核心思想:
-// - 事务内操作正常写 BufferPool(标记 dirty)
-// - Commit 时收集脏页 + WAL + fsync 落盘(保证持久化)
-// - Rollback 时仅标记回滚,实际回滚依赖 WAL 恢复机制
-//
-// 注意:
-// - 这是简化实现,适合单线程场景
-// - 多线程并发事务需要更复杂的隔离机制
+// - 事务开始时,记录所有被修改页的原始数据(快照)
+// - 事务内操作正常修改 BufferPool
+// - Commit 时将修改后的脏页写盘(WAL + fsync)
+// - Rollback 时用原始数据快照覆盖 BufferPool 中的脏页(立即生效)
 type Transaction struct {
-	db         *DB
-	committed  bool
-	rolledBack bool
+	db            *DB
+	originalPages map[uint32][]byte // pageID → 原始数据快照(用于 Rollback)
+	committed     bool
+	rolledBack    bool
 }
 
 // BeginTransaction 开启新事务
 func (db *DB) BeginTransaction() *Transaction {
 	txn := &Transaction{
-		db:         db,
-		committed:  false,
-		rolledBack: false,
+		db:            db,
+		originalPages: make(map[uint32][]byte),
+		committed:     false,
+		rolledBack:    false,
 	}
 	
 	// 设置当前活跃事务
@@ -34,6 +33,23 @@ func (db *DB) BeginTransaction() *Transaction {
 	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
+}
+
 // Commit 提交事务
 // 1. 收集所有 BufferPool 脏页
 // 2. 写入 WAL
@@ -75,13 +91,8 @@ func (txn *Transaction) Commit() error {
 	return nil
 }
 
-// Rollback 回滚事务(简化版)
-// 当前实现:仅标记回滚,不清除 BufferPool
-// 实际回滚依赖:
-// 1. 下次打开数据库时的 WAL 恢复机制
-// 2. 或者手动调用 db.RecoverFromWAL()
-//
-// TODO: 未来可以实现立即回滚(从磁盘 reload 脏页)
+// Rollback 回滚事务
+// 用原始数据快照覆盖 BufferPool 中的脏页(立即生效)
 func (txn *Transaction) Rollback() error {
 	if txn.committed {
 		return fmt.Errorf("transaction already committed, cannot rollback")
@@ -90,24 +101,22 @@ func (txn *Transaction) Rollback() error {
 		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
-	
-	// 注意:这里没有清除 BufferPool 中的脏页
-	// 如果需要立即回滚,可以调用 txn.rollbackBufferPool()
+	txn.originalPages = nil
 	
 	return nil
 }
 
-// rollbackBufferPool 从磁盘重新加载所有脏页(可选实现)
-// 这个方法可以在 Rollback 时调用,实现立即回滚
-// TODO: 需要 File 层提供 GetDirtyPages() 方法
-func (txn *Transaction) rollbackBufferPool() error {
-	// TODO: 实现立即回滚逻辑
-	return fmt.Errorf("rollbackBufferPool not yet implemented")
-}
-
 // IsCommitted 检查事务是否已提交
 func (txn *Transaction) IsCommitted() bool {
 	return txn.committed

BIN
txn_test.db