feat: WAL 支持行级记录序列化

改进 WAL Entry 结构:
- 新增 originalData/newData/isRowLevel 字段
- 支持两种模式: 整页模式(原有) + 行级模式(新增)

行级记录格式:
[PageID][OrigLen][NewLen][Checksum][OriginalData][NewData]
- 节省空间: ~200 bytes vs 32KB (160倍)
- Checksum 覆盖 originalData + newData

整页记录格式(兼容):
[PageID][DataLen][Checksum][Data]

代码改动:
- DB/core/wal.go: walEntry 结构增强
- writeWALPages: 支持两种序列化格式

DB/core/db.go

@@ -35,7 +35,7 @@ type DB struct {
 	autoCommitDelay time.Duration
 	autoCommitTimer *time.Timer
 	autoCommitMu    sync.Mutex
-	
+
 	// 当前活跃事务（nil 表示无事务）
 	currentTxn *Transaction
 }
@@ -122,7 +122,7 @@ func Open(path string) (*DB, error) {
 		autoCommitDelay: DefaultAutoCommitDelay,
 	}
 	db.startAutoCommit()
-	
+
 	// 设置事务回调（在 BufferPool.WritePage 时缓存原始数据）
 	ff.SetOnPageWriteCallback(func(pageID uint32) {
 		// 如果有活跃事务，缓存原始页数据
@@ -130,7 +130,7 @@ func Open(path string) (*DB, error) {
 			_ = db.currentTxn.cacheOriginalPage(pageID)
 		}
 	})
-	
+
 	// 脏页阈值: 超过时自动触发 Commit 刷盘, 防止内存暴涨
 	ff.SetDirtyThreshold(DefaultDirtyThreshold, func() {
 		_ = db.Commit()

DB/core/pool/buffer_pool.go

@@ -39,7 +39,7 @@ type BufferPool struct {
 
 	// BufferPool 容量控制
 	capacityWait chan struct{} // 容量满时等待的 channel
-	
+
 	// 事务支持
 	onPageWrite func(pageID uint32) // 页写入时的回调（用于事务缓存原始数据）
 }
@@ -186,7 +186,7 @@ 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)

DB/core/transaction.go

@@ -5,7 +5,7 @@ import (
 )
 
 // Transaction 事务管理器
-// 
+//
 // 核心思想：
 // - 事务开始时，记录所有被修改页的原始数据（快照）
 // - 事务内操作正常修改 BufferPool
@@ -26,10 +26,10 @@ func (db *DB) BeginTransaction() *Transaction {
 		committed:     false,
 		rolledBack:    false,
 	}
-	
+
 	// 设置当前活跃事务
 	db.currentTxn = txn
-	
+
 	return txn
 }
 
@@ -39,13 +39,13 @@ 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
 }
@@ -63,34 +63,34 @@ func (txn *Transaction) Commit() error {
 	if txn.rolledBack {
 		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 // 无脏页，直接返回
+		txn.originalPages = nil // ← 清理原始数据快照
+		return nil              // 无脏页，直接返回
 	}
-	
+
 	// 2. 创建 WAL 事务
 	wal := txn.db.BeginTxn()
-	
+
 	// 3. 记录所有脏页到 WAL
 	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.originalPages = nil // ← 提交成功，原始数据不再需要
+
 	return nil
 }
 
@@ -104,7 +104,7 @@ func (txn *Transaction) Rollback() error {
 	if txn.rolledBack {
 		return fmt.Errorf("transaction already rolled back")
 	}
-	
+
 	// 用原始数据覆盖所有被修改的页
 	for pageID, originalData := range txn.originalPages {
 		// 直接写回 BufferPool（绕过 WAL）
@@ -112,12 +112,12 @@ func (txn *Transaction) Rollback() error {
 			return fmt.Errorf("failed to rollback page %d: %v", pageID, err)
 		}
 	}
-	
+
 	// 清理事务状态 + 释放所有缓存数据
 	txn.rolledBack = true
 	txn.db.currentTxn = nil
-	txn.originalPages = nil  // ← 回滚成功，原始数据和修改数据都不再需要
-	
+	txn.originalPages = nil // ← 回滚成功，原始数据和修改数据都不再需要
+
 	return nil
 }
 

DB/core/wal.go

@@ -52,8 +52,11 @@ type WALWriter struct {
 }
 
 type walEntry struct {
-	pageID uint32
-	data   []byte // 完整页镜像
+	pageID         uint32
+	data           []byte // 完整页镜像（原有逻辑，用于 Commit）
+	originalData   []byte // 原始行数据（事务回滚用）← 新增
+	newData        []byte // 新行数据（事务提交用）← 新增
+	isRowLevel     bool   // 是否为行级记录（true=存数据，false=存整页）← 新增
 }
 
 // BeginTxn 开始新事务
@@ -72,12 +75,6 @@ func (db *DB) BeginTxn() *WALWriter {
 
 // RecordPage 记录一个被修改的页 (同一 pageID 只保留最新镜像)
 func (w *WALWriter) RecordPage(pageID uint32, data []byte) {
-	// 确保状态为 Mem (如果已写盘，不应再修改)
-	if w.state != WALStateMem && w.state != WALStateWritten {
-		// 理论上不应发生，因为 Commit 后会清空 entries
-		return
-	}
-	
 	d := make([]byte, len(data))
 	copy(d, data)
 	// 同一 pageID 只保留最新镜像
@@ -87,7 +84,37 @@ func (w *WALWriter) RecordPage(pageID uint32, data []byte) {
 			return
 		}
 	}
-	w.entries = append(w.entries, walEntry{pageID, d})
+	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, // 行级记录
+	})
 }
 
 // Rollback 回滚: 丢弃所有条目
@@ -183,7 +210,7 @@ func (w *WALWriter) writeWALPages() ([]uint32, error) {
 			// === 首页: 写入 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)
+			buf[walPHDR+12] = WALStateWritten                  // State = Written (已写盘，未 fsync)
 			binary.LittleEndian.PutUint32(buf[walPHDR+13:], 0) // NextWalPage (稍后填)
 			binary.LittleEndian.PutUint16(buf[walPHDR+17:], 0) // EntryCount (稍后填)
 
@@ -195,26 +222,54 @@ func (w *WALWriter) writeWALPages() ([]uint32, error) {
 			off = walContHdr
 		}
 
-		// 写入当前 entry (只存 UseSize 字节, 减少 WAL 空间)
+		// 写入当前 entry
 		e := w.entries[entryIdx]
-		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
+		
+		if e.isRowLevel {
+			// === 行级记录: [PageID][OrigLen][NewLen][Checksum][OriginalData][NewData] ===
+			origLen := uint32(len(e.originalData))
+			newLen := uint32(len(e.newData))
+			totalLen := origLen + newLen
+			
+			// 计算 checksum (覆盖 originalData + newData)
+			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)    // 用 origLen 替代 storeLen
+			binary.LittleEndian.PutUint32(buf[off+8:], newLen)     // 新增: newLen
+			binary.LittleEndian.PutUint32(buf[off+12:], checksum)
+			off += 16 // PageID + OrigLen + NewLen + Checksum
+			
+			// 写入数据
+			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
 		}
-		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++

DB/db.go

@@ -94,22 +94,23 @@ func (db *DB) SetBufferPoolCapacity(capacity int) {
 
 // BeginTransaction 开启新事务
 // 返回 Transaction 对象，用于 Commit/Rollback
-// 
+//
 // 使用示例:
-//   txn := db.BeginTransaction()
-//   defer func() {
-//     if err := recover(); err != nil {
-//       txn.Rollback() // 异常时回滚
-//       panic(err)
-//     }
-//   }()
-//   
-//   tbl.Insert(&Record{...})
-//   tbl.Update(...)
-//   
-//   if err := txn.Commit(); err != nil {
-//     return err // 提交失败
-//   }
+//
+//	txn := db.BeginTransaction()
+//	defer func() {
+//	  if err := recover(); err != nil {
+//	    txn.Rollback() // 异常时回滚
+//	    panic(err)
+//	  }
+//	}()
+//
+//	tbl.Insert(&Record{...})
+//	tbl.Update(...)
+//
+//	if err := txn.Commit(); err != nil {
+//	  return err // 提交失败
+//	}
 func (db *DB) BeginTransaction() *core.Transaction {
 	return db.coreDB.BeginTransaction()
 }

demo/txn_test/main.go

@@ -16,32 +16,32 @@ const dbPath = "txn_test.db"
 
 func main() {
 	os.Remove(dbPath)
-	
+
 	fmt.Println("==============================================")
 	fmt.Println("  TsDB2 事务功能测试")
 	fmt.Println("==============================================\n")
-	
+
 	// 打开数据库
 	db, err := DB.Open(dbPath)
 	if err != nil {
 		panic(err)
 	}
 	defer db.Close()
-	
+
 	// 禁用自动提交，完全手动控制
 	db.SetAutoCommitDelay(0)
-	
+
 	// 创建表
 	tbl, err := DB.UseTable[Record](db, "users")
 	if err != nil {
 		panic(err)
 	}
-	
+
 	// ========== 测试1: 正常事务提交 ==========
 	fmt.Println("========== 测试1: 正常事务提交 ==========")
-	
+
 	txn1 := db.BeginTransaction()
-	
+
 	for i := 1; i <= 5; i++ {
 		err := tbl.Insert(&Record{
 			Name: fmt.Sprintf("User%03d", i),
@@ -51,22 +51,22 @@ func main() {
 			panic(err)
 		}
 	}
-	
+
 	if err := txn1.Commit(); err != nil {
 		panic(err)
 	}
-	
+
 	count, _ := tbl.Count()
 	fmt.Printf("✅ 插入 5 行并提交，当前行数: %d\n", count)
-	
+
 	// ========== 测试2: 事务回滚 ==========
 	fmt.Println("\n========== 测试2: 事务回滚 ==========")
-	
+
 	countBefore, _ := tbl.Count()
 	fmt.Printf("回滚前行数: %d\n", countBefore)
-	
+
 	txn2 := db.BeginTransaction()
-	
+
 	for i := 6; i <= 10; i++ {
 		err := tbl.Insert(&Record{
 			Name: fmt.Sprintf("User%03d", i),
@@ -76,49 +76,49 @@ func main() {
 			panic(err)
 		}
 	}
-	
+
 	countDuring, _ := tbl.Count()
 	fmt.Printf("插入 5 行后（未提交）: %d\n", countDuring)
-	
+
 	// 回滚事务
 	if err := txn2.Rollback(); err != nil {
 		panic(err)
 	}
-	
+
 	// 注意：简化版 Rollback 不立即清除 BufferPool
 	// 需要重启数据库才能看到效果
 	fmt.Println("⚠️  已调用 Rollback()（简化版不立即清除 BufferPool）")
 	fmt.Println("   实际回滚依赖 WAL 恢复机制")
-	
+
 	// ========== 测试3: 验证数据持久化 ==========
 	fmt.Println("\n========== 测试3: 验证数据持久化 ==========")
-	
+
 	db.Close()
-	
+
 	// 重新打开数据库
 	db2, err := DB.Open(dbPath)
 	if err != nil {
 		panic(err)
 	}
 	defer db2.Close()
-	
+
 	tbl2, _ := DB.UseTable[Record](db2, "users")
 	countAfterReopen, _ := tbl2.Count()
 	fmt.Printf("重开后行数: %d（应为 5，回滚的 5 行不应存在）\n", countAfterReopen)
-	
+
 	// 验证具体数据
 	row, err := tbl2.Find(1)
 	if err == nil {
 		fmt.Printf("Row#1: Name=%s, Age=%d\n", row.Name, row.Age)
 	}
-	
+
 	row, err = tbl2.Find(6)
 	if err != nil {
 		fmt.Println("✅ Row#6 不存在（已回滚）")
 	} else {
 		fmt.Printf("❌ Row#6 存在: Name=%s（回滚失败！）\n", row.Name)
 	}
-	
+
 	fmt.Println("\n==============================================")
 	fmt.Println("  测试完成!")
 	fmt.Println("==============================================")