|
@@ -5,15 +5,16 @@ import (
|
|
|
)
|
|
)
|
|
|
|
|
|
|
|
// Transaction 事务管理器
|
|
// Transaction 事务管理器
|
|
|
-//
|
|
|
|
|
|
|
+//
|
|
|
// 核心思想:
|
|
// 核心思想:
|
|
|
-// - 事务开始时,记录所有被修改页的原始数据(快照)
|
|
|
|
|
|
|
+// - 事务开始时,记录所有被修改行的原始数据和新数据
|
|
|
// - 事务内操作正常修改 BufferPool
|
|
// - 事务内操作正常修改 BufferPool
|
|
|
-// - Commit 时将修改后的脏页写盘(WAL + fsync)
|
|
|
|
|
-// - Rollback 时用原始数据快照覆盖 BufferPool 中的脏页(立即生效)
|
|
|
|
|
|
|
+// - Commit 时将行级数据写入 WAL(节省空间)
|
|
|
|
|
+// - Rollback 时用原始数据覆盖 BufferPool 中的脏页(立即生效)
|
|
|
type Transaction struct {
|
|
type Transaction struct {
|
|
|
db *DB
|
|
db *DB
|
|
|
- originalPages map[uint32][]byte // pageID → 原始数据快照(用于 Rollback)
|
|
|
|
|
|
|
+ originalPages map[uint32][]byte // pageID → 原始页数据快照(用于 Rollback)
|
|
|
|
|
+ rowChanges map[uint32]struct{ original, new []byte } // pageID → {原始行, 新行}(用于 WAL)
|
|
|
committed bool
|
|
committed bool
|
|
|
rolledBack bool
|
|
rolledBack bool
|
|
|
}
|
|
}
|
|
@@ -23,13 +24,14 @@ func (db *DB) BeginTransaction() *Transaction {
|
|
|
txn := &Transaction{
|
|
txn := &Transaction{
|
|
|
db: db,
|
|
db: db,
|
|
|
originalPages: make(map[uint32][]byte),
|
|
originalPages: make(map[uint32][]byte),
|
|
|
|
|
+ rowChanges: make(map[uint32]struct{ original, new []byte }),
|
|
|
committed: false,
|
|
committed: false,
|
|
|
rolledBack: false,
|
|
rolledBack: false,
|
|
|
}
|
|
}
|
|
|
-
|
|
|
|
|
|
|
+
|
|
|
// 设置当前活跃事务
|
|
// 设置当前活跃事务
|
|
|
db.currentTxn = txn
|
|
db.currentTxn = txn
|
|
|
-
|
|
|
|
|
|
|
+
|
|
|
return txn
|
|
return txn
|
|
|
}
|
|
}
|
|
|
|
|
|
|
@@ -39,17 +41,27 @@ func (txn *Transaction) cacheOriginalPage(pageID uint32) error {
|
|
|
if _, exists := txn.originalPages[pageID]; exists {
|
|
if _, exists := txn.originalPages[pageID]; exists {
|
|
|
return nil
|
|
return nil
|
|
|
}
|
|
}
|
|
|
-
|
|
|
|
|
|
|
+
|
|
|
// 从磁盘读取原始数据
|
|
// 从磁盘读取原始数据
|
|
|
buf := make([]byte, PageSize)
|
|
buf := make([]byte, PageSize)
|
|
|
if err := txn.db.file.readPageDisk(pageID, buf); err != nil {
|
|
if err := txn.db.file.readPageDisk(pageID, buf); err != nil {
|
|
|
return fmt.Errorf("failed to read page %d for rollback cache: %v", pageID, err)
|
|
return fmt.Errorf("failed to read page %d for rollback cache: %v", pageID, err)
|
|
|
}
|
|
}
|
|
|
-
|
|
|
|
|
|
|
+
|
|
|
txn.originalPages[pageID] = buf
|
|
txn.originalPages[pageID] = buf
|
|
|
return nil
|
|
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 提交事务
|
|
// Commit 提交事务
|
|
|
// 1. 收集所有 BufferPool 脏页
|
|
// 1. 收集所有 BufferPool 脏页
|
|
|
// 2. 写入 WAL
|
|
// 2. 写入 WAL
|
|
@@ -64,21 +76,28 @@ func (txn *Transaction) Commit() error {
|
|
|
return fmt.Errorf("transaction already rolled back")
|
|
return fmt.Errorf("transaction already rolled back")
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
- // 1. 收集所有脏页
|
|
|
|
|
- dirty := txn.db.file.CollectDirty()
|
|
|
|
|
- if len(dirty) == 0 {
|
|
|
|
|
- txn.committed = true
|
|
|
|
|
- txn.db.currentTxn = nil
|
|
|
|
|
- txn.originalPages = nil // ← 清理原始数据快照
|
|
|
|
|
- return nil // 无脏页,直接返回
|
|
|
|
|
- }
|
|
|
|
|
-
|
|
|
|
|
- // 2. 创建 WAL 事务
|
|
|
|
|
|
|
+ // 1. 创建 WAL 事务
|
|
|
wal := txn.db.BeginTxn()
|
|
wal := txn.db.BeginTxn()
|
|
|
-
|
|
|
|
|
- // 3. 记录所有脏页到 WAL
|
|
|
|
|
- for _, d := range dirty {
|
|
|
|
|
- wal.RecordPage(d.PageID, d.Data)
|
|
|
|
|
|
|
+
|
|
|
|
|
+ // 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,确保落盘 ⭐
|
|
// 4. ⭐ 关键:Commit 会触发 fsync,确保落盘 ⭐
|
|
@@ -90,6 +109,7 @@ func (txn *Transaction) Commit() error {
|
|
|
txn.committed = true
|
|
txn.committed = true
|
|
|
txn.db.currentTxn = nil
|
|
txn.db.currentTxn = nil
|
|
|
txn.originalPages = nil // ← 提交成功,原始数据不再需要
|
|
txn.originalPages = nil // ← 提交成功,原始数据不再需要
|
|
|
|
|
+ txn.rowChanges = nil // ← 清理行级变化
|
|
|
|
|
|
|
|
return nil
|
|
return nil
|
|
|
}
|
|
}
|
|
@@ -117,6 +137,7 @@ func (txn *Transaction) Rollback() error {
|
|
|
txn.rolledBack = true
|
|
txn.rolledBack = true
|
|
|
txn.db.currentTxn = nil
|
|
txn.db.currentTxn = nil
|
|
|
txn.originalPages = nil // ← 回滚成功,原始数据和修改数据都不再需要
|
|
txn.originalPages = nil // ← 回滚成功,原始数据和修改数据都不再需要
|
|
|
|
|
+ txn.rowChanges = nil // ← 清理行级变化
|
|
|
|
|
|
|
|
return nil
|
|
return nil
|
|
|
}
|
|
}
|