Bustub MVCC

TableHeap（最新版本，通常是正在进行中的事务或最后一次提交的结果）
    ↓  (通过 TransactionManager 中的 version_info_ 映射)
VersionUndoLink { prev_: UndoLink, in_progress_: bool }
    ↓  prev_
UndoLog₁ { ts_, is_deleted_, modified_fields_, tuple_, prev_version_ }
    ↓  prev_version_
UndoLog₂
    ↓
UndoLog₃ → INVALID (链尾)

// src/include/storage/table/tuple.h
struct TupleMeta {
  timestamp_t ts_;      // 双重含义：
                        //   < TXN_START_ID → 已提交事务的 commit_ts
                        //   >= TXN_START_ID → 正在进行事务的临时 txn_id
  bool is_deleted_;     // 软删除标志（逻辑删除）
};
static constexpr size_t TUPLE_META_SIZE = 16;  // 固定 16 字节

// src/include/concurrency/transaction.h
struct UndoLink {
  txn_id_t prev_txn_;       // 指向拥有该 UndoLog 的事务 ID
  int32_t prev_log_idx_;    // 该 UndoLog 在事务 undo_logs_ 数组中的下标

  auto IsValid() const -> bool { ... }  // prev_txn_ != INVALID_TXN_ID
};

struct UndoLog {
  bool is_deleted_;                      // 这一步操作是否是删除
  std::vector<bool> modified_fields_;    // 位图：哪些列在这一步被修改了
  Tuple tuple_;                          // 只存 modified_fields_ 为 true 的列（稀疏存储）
  timestamp_t ts_;                       // 这条 UndoLog 对应的事务提交时间
  UndoLink prev_version_;               // 指向更旧版本的 UndoLog
};

// src/include/concurrency/transaction_manager.h
struct VersionUndoLink {
  UndoLink prev_;           // 指向最新 UndoLog 的链接
  bool in_progress_;        // 当前是否有事务正在修改此 tuple

  static auto FromOptionalUndoLink(std::optional<UndoLink> undo_link) -> VersionUndoLink;
};

// src/include/concurrency/transaction_manager.h
struct PageVersionInfo {
  std::shared_mutex mutex_;   // 保护此 Page 内所有 slot 的版本链头
  std::unordered_map<slot_offset_t, VersionUndoLink> prev_version_;
};

// TransactionManager 成员
std::unordered_map<page_id_t, std::shared_ptr<PageVersionInfo>> version_info_;

// src/include/common/config.h
static constexpr txn_id_t TXN_START_ID = 1LL << 62;  // 极大的数

事务 Begin 时：read_ts = last_commit_ts（当前最新提交的时间戳）
事务 Commit 时：commit_ts = ++last_commit_ts（原子递增获取新时间戳）

auto TransactionManager::Begin(IsolationLevel isolation_level) -> Transaction * {
  std::unique_lock<std::shared_mutex> l(txn_map_mutex_);
  auto txn_id = next_txn_id_++;
  auto txn = std::make_unique<Transaction>(txn_id, isolation_level);
  auto *txn_ref = txn.get();
  txn_map_.insert(std::make_pair(txn_id, std::move(txn)));

  // 关键：read_ts = 当前最新提交时间戳（原子读取）
  txn_ref->read_ts_ = last_commit_ts_.load();

  // 注册到 Watermark 追踪器
  running_txns_.AddTxn(txn_ref->read_ts_);
  return txn_ref;
}

活跃事务：T1(read_ts=5), T2(read_ts=7), T3(read_ts=9)
Watermark = min(5, 7, 9) = 5

auto Watermark::AddTxn(timestamp_t read_ts) -> void {
  if (read_ts < commit_ts_) {
    throw Exception("read ts < commit ts");
  }
  // 若 map 为空（没有活跃事务），直接设置 watermark
  if (current_reads_.empty()) {
    watermark_ = read_ts;
  }
  ++current_reads_[read_ts];  // 频率计数
}

auto Watermark::RemoveTxn(timestamp_t read_ts) -> void {
  auto iter = current_reads_.find(read_ts);
  if (iter == current_reads_.end()) {
    throw Exception("read ts not found");
  }
  if (iter->second == 1) {
    current_reads_.erase(iter);   // 最后一个使用此 read_ts 的事务退出
  } else {
    --iter->second;               // 还有其他事务使用相同 read_ts
  }
  // 若移除的是当前 watermark，需要更新 watermark
  if (read_ts == watermark_ && !current_reads_.empty()) {
    // 向前推进 watermark，找到下一个有活跃事务的 read_ts
    // current_reads_ 是有序 map，可直接取 begin()->first，无需循环
  }
}

ts_ 值域分析：

[0, TXN_START_ID)      → 已提交事务的 commit_ts（小整数）
[TXN_START_ID, MAX)    → 进行中事务的临时 txn_id（极大整数）

令 tuple_ts = tuple_meta.ts_
令 txn_ts = txn->GetReadTs()
令 my_txn_id = txn->GetTransactionId()  // >= TXN_START_ID

情况1: tuple_ts <= txn_ts
    → tuple 已提交且时间 ≤ 当前快照，直接可见

情况2: tuple_ts == my_txn_id
    → tuple 是本事务自己修改的，可见（读自己写的数据）

情况3: tuple_ts > txn_ts && tuple_ts != my_txn_id
    → tuple 被一个更新的已提交事务或另一个未提交事务修改
    → 需要沿 undo log 链向后找到对当前快照可见的版本
    → 可能的情况：
       3a: tuple_ts >= TXN_START_ID（另一个正在进行的事务修改）
       3b: tuple_ts < TXN_START_ID 但 > txn_ts（另一个较新的已提交事务修改）

struct UndoLog {
  bool is_deleted_;                   // true 表示"这步操作是删除"
  std::vector<bool> modified_fields_; // 大小 = schema.GetColumnCount()
                                      // modified_fields_[i] = true → 第 i 列在此 log 中有旧值
  Tuple tuple_;                       // 只包含 modified_fields_ 为 true 的列
                                      // 用 Schema::CopySchema() 生成子 schema 来解析
  timestamp_t ts_;                    // 此 UndoLog 记录的版本对应的时间戳
  UndoLink prev_version_;             // 指向更旧的 UndoLog
};

modified_fields_ = [false, true, false, true]
tuple_.data_      = [(b=2), (d=4)]  // 只存修改了的列的旧值
                    // 解析时用 CopySchema([1, 3]) 生成 2 列 schema

TableHeap tuple: ts_=TXN8（事务8临时id，表示事务8正在修改）
    ↓
UndoLog1: ts_=5（表示"若你想看 commit_ts=5 时的状态，apply 这个 log"）
    ↓
UndoLog2: ts_=3
    ↓
INVALID（链尾，再往前没有版本记录）

auto ReconstructTuple(const Schema *schema,
                      const Tuple &base_tuple,    // TableHeap 中的最新 tuple
                      const TupleMeta &base_meta, // 对应的元数据
                      const std::vector<UndoLog> &undo_logs  // 从新到旧的 undo log 数组
) -> std::optional<Tuple>;
// 返回 nullopt 表示重建后的版本已被删除（即该历史版本不存在）

base_tuple (TableHeap 当前版本)
    │
    ▼ apply UndoLog₁ (最近的变更 → 还原这一步)
 版本v₁
    │
    ▼ apply UndoLog₂
 版本v₂ ← 这可能就是我们需要的历史版本
    │
    ▼ apply UndoLog₃
 版本v₃（更古老的版本）

新 ←————————————————→ 旧

TableHeap → UndoLog1 → UndoLog2 → UndoLog3 → INVALID
            (最近的变更)          (最早的变更)

// SeqScan 中的使用（seq_scan_executor.cpp）
std::vector<UndoLog> undo_logs;
UndoLink undo_link = txn_manager->GetUndoLink(*rid).value_or(UndoLink{});
auto optional_undo_log = txn_manager->GetUndoLogOptional(undo_link);

// 收集足够多的 undo log，直到时间戳满足条件
while (optional_undo_log.has_value() && tuple_ts > txn_ts) {
  undo_logs.push_back(*optional_undo_log);
  tuple_ts = optional_undo_log->ts_;
  undo_link = optional_undo_log->prev_version_;
  optional_undo_log = txn_manager->GetUndoLogOptional(undo_link);
}

// 检查是否找到了满足条件的版本
if (tuple_ts > txn_ts) {
  continue;  // 整个版本链都比 txn_ts 新，此 tuple 对当前事务不可见
}

auto new_tuple = ReconstructTuple(&schema, *tuple, tuple_meta, undo_logs);

auto InsertExecutor::Next([[maybe_unused]] Tuple *tuple, RID *rid) -> bool {
  if (has_inserted_) { return false; }  // 确保只执行一次
  has_inserted_ = true;

  int count = 0;
  auto txn = exec_ctx_->GetTransaction();
  auto txn_mgr = exec_ctx_->GetTransactionManager();
  auto catalog = exec_ctx_->GetCatalog();
  auto table_info = catalog->GetTable(plan_->GetTableOid());
  auto indexes = catalog->GetTableIndexes(table_info->name_);

  while (child_executor_->Next(tuple, rid)) {
    InsertTuple(txn, txn_mgr, table_info, indexes, *tuple);
    ++count;
  }
  // 输出受影响行数
  std::vector<Value> result{{TypeId::INTEGER, count}};
  *tuple = Tuple{result, &GetOutputSchema()};
  return true;
}

auto TransactionManager::Commit(Transaction *txn) -> bool {
  std::unique_lock<std::mutex> commit_lck(commit_mutex_);  // 串行化提交
  // commit_ts = ++last_commit_ts_
  // 遍历 write_set_，将所有相关 tuple 的 ts_ 改为 commit_ts
  // running_txns_.RemoveTxn(txn->read_ts_)
  return true;
}

// 计算哪些列发生了变化
std::vector<bool> new_modified_fields;
std::vector<Value> new_modified_values;
std::vector<uint32_t> cols;
for (uint32_t i = 0; i < schema.GetColumnCount(); ++i) {
  auto old_value = cur_tuple.GetValue(&schema, i);
  if (old_value.CompareEquals(new_values[i]) == CmpBool::CmpTrue) {
    new_modified_fields.emplace_back(false);  // 未变化
  } else {
    new_modified_fields.emplace_back(true);   // 已变化，记录旧值
    new_modified_values.push_back(old_value);
  }
}
// 创建 UndoLog 并追加到事务的 undo_logs_ 数组
undo_link = txn->AppendUndoLog(UndoLog{
    tuple_meta.is_deleted_,
    new_modified_fields,
    new_modified_tuple,
    tuple_meta.ts_,  // 旧版本的时间戳
    undo_link        // 指向更旧的版本
});

// 目标：undo_log 始终记录"从事务开始前的状态到当前状态的完整差异"

// 事务第一次写：a=1→2  → undo_log: {a: 1}（记录 a 的原始值）
// 事务第二次写：a=2→3, b=5→6  → undo_log 合并为: {a: 1, b: 5}
//            （a 的原始值仍是 1，b 新加入，原始值是 5）

// 若不合并而是新建：
//   undo_log1: {a: 2, b: 5} → undo_log0: {a: 1}
//   回滚时需要 apply 两次，链条更长
txn->ModifyUndoLog(undo_link.prev_log_idx_, UndoLog{
    undo_log->is_deleted_,
    merged_modified_fields,
    merged_modified_tuple,
    undo_log->ts_,
    undo_log->prev_version_
});

阶段一（收集）: while(child->Next()) → old_tuples, new_tuples, rids 全部收集
阶段二（写入）: 遍历收集好的数据，执行实际的写操作

提交状态: (deleted, ts=5) → UndoLog₁(is_deleted=false, ..., ts=3) → INVALID
                                      ↑ 旧的有效数据

插入后:   (value=X, ts=TXN8) → UndoLog_new(is_deleted=true, ts=5) → UndoLog₁ → INVALID
                              ↑ 新插入的 UndoLog 记录"我是在删除状态上覆盖的"

场景：两个事务同时插入相同主键

T1 begin(read_ts=5):
  InsertTuple(key=X) → 索引插入: key=X → RID=1, tuple_ts=TXN1

T2 begin(read_ts=5):（与 T1 并发）
  InsertTuple(key=X)
  → ScanKey 找到 RID=1，tuple_ts=TXN1 > read_ts=5 && != T2的id → 写写冲突 → abort T2

void TransactionManager::Abort(Transaction *txn) {
  std::unique_lock<std::mutex> commit_lck(commit_mutex_);
  // 遍历 write_set_
  // 对每行：通过第一 UndoLog 还原，或直接删除（新插入行）
  running_txns_.RemoveTxn(txn->read_ts_);
}

事务时间线：
    T1 begin(read_ts=5) ─────────────── T1 commit
    T2 begin ──────── T2 commit(commit_ts=7)

T2 的 write set: {row_A}
T1 的 scan predicates: {table=employees, predicate=(salary < 10000)}

VerifyTxn 检测：
  row_A 在 commit_ts=7 时是否满足 (salary < 10000)?
  如果满足 → T1 扫描时"本应"看到这行变化 → 幻读 → abort T1

Begin()
  └── read_ts = last_commit_ts（原子读）
  └── txn_id = next_txn_id_++
  └── 加入 txn_map_，注册到 Watermark

执行阶段（Read/Write）
  └── Read: 可见性检查 → 按需 ReconstructTuple
  └── Write: 冲突检测 → 更新 TableHeap → 创建/合并 UndoLog → 记录 write_set_

Commit()
  └── [commit_mutex_] commit_ts = ++last_commit_ts
  └── write_set_ 中所有 tuple: ts_ = commit_ts
  └── VersionLink.in_progress = false（释放"写锁"标记）
  └── running_txns_.RemoveTxn(read_ts)

Abort()
  └── 遍历 write_set_，通过 UndoLog 还原所有修改
  └── 无 UndoLog 的 tuple → is_deleted=true, ts_=0
  └── running_txns_.RemoveTxn(read_ts)

GarbageCollection()
  └── watermark = running_txns_.GetWatermark()
  └── 从 txn_map_ 删除 commit_ts < watermark 的已完成事务

初始状态（commit_ts=2）:
  TableHeap: (a=1, b=5), ts_=2, is_deleted_=false
  version_info_: prev_ = INVALID

T1(read_ts=2) 执行 UPDATE a=10:
  创建 UndoLog: {is_deleted_=false, modified_fields_=[true,false], tuple_=(a=1), ts_=2, prev_=INVALID}
  TableHeap: (a=10, b=5), ts_=TXN1
  version_info_: prev_ → UndoLog_T1

T1 Commit (commit_ts=5):
  TableHeap: (a=10, b=5), ts_=5

T2(read_ts=5) 执行 DELETE:
  创建 UndoLog: {is_deleted_=false, modified_fields_=[true,true], tuple_=(a=10,b=5), ts_=5, prev_→UndoLog_T1}
  TableHeap: ts_=TXN2, is_deleted_=true
  version_info_: prev_ → UndoLog_T2 → UndoLog_T1

T2 Commit (commit_ts=8):
  TableHeap: (a=10, b=5), ts_=8, is_deleted_=true

T3(read_ts=4) 读这行:
  tuple_ts=8 > read_ts=4 → 需要重建
  collect: UndoLog_T2(ts_=5>4) → UndoLog_T1(ts_=2≤4), 停止
  apply UndoLog_T2: 撤销 T2 的删除 → (a=10, b=5, deleted=false)
  结果: (a=10, b=5) 可见（T1 的修改对 T3 可见，T2 的删除对 T3 不可见）