TsDB2/DB/core/transaction.go

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
}