BusTub Query Execution

SQL 字符串
    ↓  Parser（词法 + 语法分析）
抽象语法树 AST
    ↓  Binder（语义绑定，解析表名/列名）
Bound AST（带 Schema 信息）
    ↓  Planner（生成逻辑计划）
LogicalPlanNode 树
    ↓  Optimizer（规则/代价优化）
PhysicalPlanNode 树（AbstractPlanNodeRef）
    ↓  ExecutorFactory（为每个 PlanNode 创建对应 Executor）
Executor 树
    ↓  根 Executor 调用 Next()，火山模型驱动执行
结果 Tuples

EXPLAIN SELECT * FROM t1 WHERE id = 1;
-- 输出类似：
-- === OPTIMIZER ===
-- IndexScan { index_oid=0, filter=... } | (id:INTEGER, val:INTEGER)

class AbstractExecutor {
 public:
  explicit AbstractExecutor(ExecutorContext *exec_ctx) : exec_ctx_{exec_ctx} {}
  virtual ~AbstractExecutor() = default;

  // 必须在 Next() 之前调用，初始化算子（递归初始化子算子）
  virtual void Init() = 0;

  // 返回下一条 tuple；true = 有数据，false = 数据已耗尽
  virtual auto Next(Tuple *tuple, RID *rid) -> bool = 0;

  // 返回该算子输出的 Schema
  virtual auto GetOutputSchema() const -> const Schema & = 0;

 protected:
  ExecutorContext *exec_ctx_;
};

根节点 ProjectionExecutor
    ↑  Next() — 拉取下一条 tuple
    ↓  调用子算子 Next()
FilterExecutor
    ↑  Next()
    ↓  调用子算子 Next()
SeqScanExecutor
    ↑  Next()
    ↓  读取 TableHeap
TableHeap / BufferPool

用户发起查询
    │
    ▼
ExecutorEngine::Execute(plan)
    │
    ├─ ExecutorFactory::CreateExecutor(plan)  // 递归创建 Executor 树
    │
    ├─ root_executor->Init()                  // 递归初始化
    │
    └─ while (root_executor->Next(&tuple, &rid)):
           results.push_back(tuple)

AbstractPlanNode（只描述"做什么"）
    ├─ 持有 output_schema_（输出 Schema）
    ├─ 持有 children_（子计划节点）
    └─ 只含元数据，不执行任何 I/O

AbstractExecutor（实际执行"怎么做"）
    ├─ 持有 PlanNode 指针（读取计划参数）
    ├─ 持有 child executors（递归执行）
    └─ 实际操作 TableHeap、Index、Buffer Pool

enum class PlanType {
  SeqScan, IndexScan, Insert, Update, Delete,
  Aggregation, Limit, NestedLoopJoin, NestedIndexJoin,
  HashJoin, Filter, Values, Projection, Sort,
  TopN, TopNPerGroup, MockScan, InitCheck, Window
};

class ExecutorContext {
 public:
  auto GetCatalog() -> Catalog *;            // 系统目录（表/索引元数据）
  auto GetTransaction() -> Transaction *;    // 当前事务
  auto GetTransactionManager() -> TransactionManager *;  // MVCC 事务管理
  auto GetBufferPoolManager() -> BufferPoolManager *;    // 缓冲池
  auto GetLockManager() -> LockManager *;    // 锁管理器（2PL）
};

auto ExecutorFactory::CreateExecutor(ExecutorContext *exec_ctx,
                                     const AbstractPlanNodeRef &plan)
    -> std::unique_ptr<AbstractExecutor> {
  switch (plan->GetType()) {
    case PlanType::SeqScan:
      return std::make_unique<SeqScanExecutor>(exec_ctx, ...);
    case PlanType::HashJoin: {
      auto left = CreateExecutor(exec_ctx, plan->GetChildAt(0));  // 递归
      auto right = CreateExecutor(exec_ctx, plan->GetChildAt(1));
      return std::make_unique<HashJoinExecutor>(exec_ctx, ..., std::move(left), std::move(right));
    }
    // ...
  }
}

Schema: (id:INTEGER, name:VARCHAR(32), age:INTEGER)
Tuple:  [4字节: id值] [变长: name值] [4字节: age值]

class Schema {
  std::vector<Column> columns_;  // 每列的名称、类型、偏移量
  static auto CopySchema(const Schema *from, const std::vector<uint32_t> &attrs) -> Schema;
};

void SeqScanExecutor::Init() {
  // 创建 TableHeap 迭代器
  table_iterator_ = std::make_unique<TableIterator>(
      exec_ctx_->GetCatalog()->GetTable(plan_->GetTableOid())->table_->MakeIterator());
}

auto SeqScanExecutor::Next(Tuple *tuple, RID *rid) -> bool {
  // 循环直到找到一条对当前事务可见的有效 tuple
  while (!table_iterator_->IsEnd()) {
    *rid = table_iterator_->GetRID();
    // 加表页读锁
    auto page_guard = table_info->table_->AcquireTablePageReadLock(*rid);
    auto page = page_guard.As<TablePage>();
    auto [tuple_meta, tuple_data] = table_info->table_->GetTupleWithLockAcquired(*rid, page);
    *tuple = tuple_data;
    ++(*table_iterator_);

    // MVCC 可见性判断（详见第 10 节）
    auto tuple_ts = tuple_meta.ts_;
    auto txn_ts = txn->GetReadTs();
    if (tuple_ts > txn_ts && tuple_ts != txn->GetTransactionId()) {
      // 需要沿 undo log 链回溯重建历史版本
      // ...
    }

    // 谓词过滤
    if (!is_deleted && 谓词满足) {
      return true;
    }
  }
  return false;
}

// 在 SeqScan 内部过滤，减少传给上层算子的数据量
if (!is_deleted &&
    !(plan_->filter_predicate_ != nullptr &&
      !plan_->filter_predicate_->Evaluate(tuple, schema).GetAs<bool>())) {
  return true;
}

// 问题：每次 Next() 调用时都重新查找 Catalog
auto table_info = exec_ctx_->GetCatalog()->GetTable(plan_->GetTableOid());
// 优化：应在 Init() 中缓存 table_info 指针

TableHeap
  ├── Page 0 (TablePage)
  │     ├── Header (free space pointer, slot count, ...)
  │     ├── Slot 0 → Tuple data
  │     ├── Slot 1 → Tuple data (deleted)
  │     └── Slot 2 → Tuple data
  ├── Page 1 (TablePage)
  │     └── ...
  └── ...

auto InsertExecutor::Next([[maybe_unused]] Tuple *tuple, RID *rid) -> bool {
  if (has_inserted_) { return false; }  // 幂等保护
  has_inserted_ = true;

  int count = 0;
  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;
}

InsertTuple(txn, txn_mgr, table_info, indexes, tuple):

Step 1: 检查所有索引是否存在冲突
  for each index:
    ScanKey(tuple.key) → 若找到 RID：
      若对应 tuple 未删除 → 唯一键冲突，abort 事务
      若对应 tuple 已删除但被其他进行中的事务操作 → 写-写冲突，abort 事务

Step 2A: 若无冲突（全新插入）:
  TableHeap::InsertTuple() → 分配新 RID
  UpdateVersionLink(new_rid, VersionUndoLink{}, nullptr)  // 初始化版本链
  AppendWriteSet(table_oid, new_rid)                      // 记录写集合（用于 abort 时回滚）
  for each index: InsertEntry(key, new_rid)               // 更新所有索引

Step 2B: 若找到已删除的同键 tuple（重复利用旧 RID）:
  创建 UndoLog 记录删除状态（用于事务回滚）
  UpdateTupleInPlace() 写入新数据，同时更新 ts_ = txn->GetTransactionTempTs()

Phase 1（收集）: 扫描所有待更新的 tuple，暂存到内存
  while child_executor_->Next():
    计算 new_values (by target_expressions_)
    old_tuples.push_back(*tuple)
    new_tuples.push_back(new_tuple)
    rids.push_back(*rid)

Phase 2（写入）: 判断是否涉及索引列，选择更新策略
  if has_index_update:
    for each rid: DeleteTuple()   // 先全部删除
    for each new_tuple: InsertTuple()  // 再全部插入
  else:
    for each (rid, new_tuple): InPlaceUpdate() + 维护 UndoLog

1. 对比 cur_tuple 与 new_tuple 的每列值
2. 生成 modified_fields[] 位图（哪些列变了）
3. 生成 modified_values[] 存储旧值
4. 创建 UndoLog{is_deleted=false, modified_fields, modified_tuple, ts=tuple_meta.ts_, prev=old_undo_link}
5. undo_link = txn->AppendUndoLog(undo_log)        // 追加到事务 undo log 数组
6. txn_mgr->UpdateVersionLink(rid, {undo_link, true})  // 更新版本链头
7. txn->AppendWriteSet(table_oid, rid)              // 加入写集
8. UpdateTupleInPlace(new_tuple) 写入最新数据

// 目标：合并 undo log，保证 undo 操作始终能还原到事务开始时的原始状态
1. 取出当前 undo_log（记录的是第一次修改相对于原始状态的差异）
2. 通过 ReconstructTuple() 重建原始 tuple（事务开始时的状态）
3. 对比 new_values 与 original_tuple 的差异，生成新的 modified_fields
4. txn->ModifyUndoLog(idx, new_undo_log)   // 原地更新已有 undo log

// 检查 target_expressions_ 中是否有非 ColumnValueExpression 的表达式（即有实际计算）
for (uint32_t col_idx = 0; col_idx < schema.GetColumnCount(); ++col_idx) {
  auto *column_value_expr = dynamic_cast<const ColumnValueExpression *>(expr.get());
  if (column_value_expr == nullptr) {  // 该列有实际修改
    for (auto index : indexes) {
      if (该索引包含 col_idx) {
        has_index_update = true;
      }
    }
  }
}

auto DeleteExecutor::Next([[maybe_unused]] Tuple *tuple, RID *rid) -> bool {
  if (has_deleted_) { return false; }
  has_deleted_ = true;

  int count = 0;
  while (child_executor_->Next(tuple, rid)) {
    DeleteTuple(txn, txn_mgr, table_info, *rid);
    ++count;
  }
  // 返回受影响行数
  *tuple = Tuple{{TypeId::INTEGER, count}, &GetOutputSchema()};
  return true;
}

DeleteTuple(txn, txn_mgr, table_info, rid):

加 TablePage 写锁
读取 tuple_meta 和当前 tuple 数据

Case 1: tuple_meta.ts_ <= txn->GetReadTs()  （已提交事务的数据）
  → 创建 UndoLog{is_deleted=false, all_fields=true, 完整旧 tuple, ts=原始ts, prev=old_link}
  → UpdateVersionLink(VersionUndoLink{new_link, true})
  → AppendWriteSet(rid)
  → page->UpdateTupleMeta({txn_temp_ts, is_deleted=true})  // 标记删除

Case 2: tuple_meta.ts_ == txn->GetTransactionId()  （本事务已修改过）
  → 若上次操作是非删除的修改：
    取出已有 undo_log，重建原始 tuple
    修改 undo_log 中的 modified_fields 为全 true，记录完整旧值
    ModifyUndoLog() 原地更新
  → 将 tuple 标记 is_deleted=true（如果之前是 Insert，则直接变成"从未存在"）

Case 3: tuple_meta.ts_ > txn->GetReadTs()  （写-写冲突）
  → txn->SetTainted() + throw ExecutionException("write-write conflict")

auto IndexScanExecutor::Next(Tuple *tuple, RID *rid) -> bool {
  if (has_scanned_) { return false; }  // 点查只执行一次
  has_scanned_ = true;

  // 从计划节点获取查询键值
  auto value = plan_->pred_key_->val_;
  std::vector<Value> values{value};
  Tuple index_key(values, &key_schema);

  // 哈希索引点查，获取 RID 列表
  std::vector<RID> rids;
  auto htable = dynamic_cast<HashTableIndexForTwoIntegerColumn *>(index_info->index_.get());
  htable->ScanKey(index_key, &rids, exec_ctx_->GetTransaction());

  if (rids.empty()) { return false; }

  *rid = rids[0];
  // 读取 tuple + MVCC 可见性判断（与 SeqScan 逻辑完全相同）
  // ...
}

AbstractExpression（抽象基类）
  │   virtual Value Evaluate(const Tuple *, const Schema &) const = 0
  │   virtual Value EvaluateJoin(const Tuple *left, const Schema &, ...) const = 0
  │
  ├── ColumnValueExpression   // 取某列的值，如 #0.1（第 0 个 child 的第 1 列）
  ├── ConstantValueExpression // 常量，如 42, 'hello'
  ├── ComparisonExpression    // 比较，如 a = 1, b > 2
  ├── ArithmeticExpression    // 算术，如 a + b, a * 2
  ├── LogicExpression         // 逻辑，如 a AND b, a OR b
  └── AggregateValueExpression // 引用聚合结果

LogicExpression(AND)
  ├── ComparisonExpression(=)
  │     ├── ColumnValueExpression(col=0)  // id 列
  │     └── ConstantValueExpression(1)
  └── ComparisonExpression(>)
        ├── ColumnValueExpression(col=2)  // age 列
        └── ConstantValueExpression(18)

// 以 ComparisonExpression 为例
auto Value = ComparisonExpression::Evaluate(tuple, schema) {
  auto left = GetChildAt(0)->Evaluate(tuple, schema);
  auto right = GetChildAt(1)->Evaluate(tuple, schema);
  return left.CompareEquals(right) == CmpBool::CmpTrue
             ? ValueFactory::GetBooleanValue(true)
             : ValueFactory::GetBooleanValue(false);
}

// FilterExecutor 中
auto value = filter_expr->Evaluate(tuple, schema);
if (!value.IsNull() && value.GetAs<bool>()) {  // NULL 不通过过滤
  return true;
}

class SimpleAggregationHashTable {
  std::unordered_map<AggregateKey, AggregateValue> ht_;
  // AggregateKey  = {group_by_values}，如 (dept_id, location) 是 GROUP BY 的键
  // AggregateValue = {aggregate_results}，如 (count, sum, min, max) 是聚合结果
};

// 通过 std::hash<AggregateKey> 特化实现
// 对多列键做组合哈希（Hash Combine）
size_t curr_hash = 0;
for (const auto &val : key.group_bys_) {
  curr_hash = HashUtil::CombineHashes(curr_hash, HashUtil::HashValue(&val));
}

void CombineAggregateValues(AggregateValue *result, const AggregateValue &input) {
  for (uint32_t i = 0; i < agg_exprs_.size(); i++) {
    switch (agg_types_[i]) {
      case CountStarAggregate:  // COUNT(*): +1，不管输入值
        result->aggregates_[i] = result->aggregates_[i].Add(Value{INTEGER, 1});
        break;
      case CountAggregate:      // COUNT(col): 非 NULL 才 +1
        if (!input.aggregates_[i].IsNull()) {
          result->aggregates_[i] = IsNull ? 1 : result + 1;
        }
        break;
      case SumAggregate:        // SUM: 非 NULL 才加
        if (!input.aggregates_[i].IsNull()) {
          result->aggregates_[i] = IsNull ? input : result + input;
        }
        break;
      case MinAggregate:        // MIN: 非 NULL 且更小才更新
        if (!input.IsNull() && (result.IsNull() || result > input)) {
          result->aggregates_[i] = input;
        }
        break;
      case MaxAggregate:        // MAX: 非 NULL 且更大才更新
        // ... 类似 MIN，方向相反
        break;
    }
  }
}

void AggregationExecutor::Init() {
  child_executor_->Init();
  aht_ = std::make_unique<SimpleAggregationHashTable>(...);

  // ★ Pipeline Breaker：在 Init() 中消费所有子算子数据
  Tuple tuple{}; RID rid{};
  while (child_executor_->Next(&tuple, &rid)) {
    aht_->InsertCombine(MakeAggregateKey(&tuple), MakeAggregateValue(&tuple));
  }
  aht_iterator_ = std::make_unique<Iterator>(aht_->Begin());
}

auto AggregationExecutor::Next(Tuple *tuple, RID *rid) -> bool {
  // Init() 已完成全部聚合，Next() 只需遍历哈希表
  if (*aht_iterator_ != aht_->End()) {
    // 拼接 group_by_keys + aggregate_values 输出
    ++(*aht_iterator_);
    return true;
  }
  // 空表特殊处理
  if (!has_executed_ && plan_->group_bys_.empty()) {
    has_executed_ = true;
    // SELECT COUNT(*) FROM empty_table → 返回 (0)，而非空结果
    return true;
  }
  return false;
}

对外表（左表）的每一行 r:
  对内表（右表）的每一行 s:
    if JoinCondition(r, s):
      输出 (r, s)

// 连接键：多列值组成的复合键
struct HashJoinKey {
  std::vector<Value> values_;
  auto operator==(const HashJoinKey &other) const -> bool { /* 逐列比较 */ }
};

// 自定义哈希函数（注册到 std::hash）
namespace std {
  template <>
  struct hash<bustub::HashJoinKey> {
    auto operator()(const HashJoinKey &key) const -> std::size_t {
      size_t curr_hash = 0;
      for (const auto &val : key.values_) {
        if (!val.IsNull()) {
          curr_hash = HashUtil::CombineHashes(curr_hash, HashUtil::HashValue(&val));
        }
      }
      return curr_hash;
    }
  };
}

// 哈希表：key → 右表中匹配的所有行（支持一对多）
std::unordered_map<HashJoinKey, std::vector<HashJoinValue>> hash_table_;

// Build 阶段（在 Init() 中完成）
void HashJoinExecutor::Init() {
  right_child_->Init();
  while (right_child_->Next(&tuple, &rid)) {
    auto key = MakeHashJoinKey(&tuple, right_expressions, right_schema);
    hash_table_[key].push_back(MakeHashJoinValue(&tuple, right_schema));
  }
}

// Probe 阶段（在 Next() 中流式进行）
auto HashJoinExecutor::Next(Tuple *tuple, RID *rid) -> bool {
  // 先消耗已有缓存结果
  if (!results_.empty()) {
    *tuple = results_.front();
    results_.pop_front();
    return true;
  }

  // 继续探测左表
  while (left_child_->Next(tuple, rid)) {
    auto key = MakeHashJoinKey(tuple, left_expressions, left_schema);
    if (hash_table_.find(key) != hash_table_.end()) {
      // 找到匹配：右表可能有多行匹配，全部放入 results_ 缓冲
      for (auto &right_val : hash_table_[key]) {
        results_.push_back(合并左右tuple);
      }
    } else if (plan_->GetJoinType() == JoinType::LEFT) {
      // LEFT JOIN：右侧填 NULL
      results_.push_back(左表行 + NULL...);
    }

    if (!results_.empty()) {
      *tuple = results_.front();
      results_.pop_front();
      return true;
    }
  }
  return false;  // 左表耗尽
}

Left tuple A 匹配 Right 中的 [B1, B2, B3]
results_ = [(A,B1), (A,B2), (A,B3)]

Next() call 1: 返回 (A,B1)，results_ = [(A,B2), (A,B3)]
Next() call 2: 返回 (A,B2)，results_ = [(A,B3)]
Next() call 3: 返回 (A,B3)，results_ = []
Next() call 4: 继续读左表...

// 优化前
std::list<Tuple> results_;  // pop_front() 有内存分配
// 优化后
std::vector<Tuple> results_;
size_t result_idx_ = 0;  // 直接移动索引，避免内存分配

void SortExecutor::Init() {
  child_executor_->Init();
  tuples_.clear();

  // ★ Pipeline Breaker：必须消费所有 tuple 才能开始排序
  Tuple tuple; RID rid;
  while (child_executor_->Next(&tuple, &rid)) {
    tuples_.emplace_back(tuple);
  }

  // 多列排序（ORDER BY col1 ASC, col2 DESC, ...）
  std::sort(tuples_.begin(), tuples_.end(),
    [this](const Tuple &lhs, const Tuple &rhs) {
      for (const auto &order_by : plan_->GetOrderBy()) {
        auto order_by_type = order_by.first;
        auto expr = order_by.second.get();
        auto lhs_val = expr->Evaluate(&lhs, plan_->OutputSchema());
        auto rhs_val = expr->Evaluate(&rhs, plan_->OutputSchema());

        if (lhs_val.CompareLessThan(rhs_val) == CmpBool::CmpTrue) {
          return order_by_type != OrderByType::DESC;  // ASC: lhs < rhs → true
        }
        if (lhs_val.CompareGreaterThan(rhs_val) == CmpBool::CmpTrue) {
          return order_by_type == OrderByType::DESC;  // DESC: lhs > rhs → true
        }
        // 相等则继续比较下一列
      }
      return true;
    });
  curr_pos_ = 0;
}

Phase 1 (Run Generation):
  将数据按 Buffer Pool 大小分块读入
  在内存中排序每块 → 生成有序的 Run 文件写入磁盘

Phase 2 (Merge):
  使用 K 路归并（K-way Merge）合并所有 Run 文件
  每次从每个 Run 的头部取最小值（最大堆辅助）
  → 输出全局有序数据

auto LimitExecutor::Next(Tuple *tuple, RID *rid) -> bool {
  if (count_ >= limit_) { return false; }  // 达到上限，终止
  if (child_executor_->Next(tuple, rid)) {
    count_++;
    return true;
  }
  return false;
}

struct Compare {
  explicit Compare(const TopNPlanNode *plan) : plan_(plan) {}

  // 返回 true 意味着 lhs "更差"（应该被堆 pop 掉）
  // 对于 ASC 排序取前 N 小：使用最大堆，"更差" = 更大
  auto operator()(const Tuple &lhs, const Tuple &rhs) const -> bool {
    for (const auto &order_by : plan_->GetOrderBy()) {
      auto order_by_type = order_by.first;
      auto lhs_val = expr->Evaluate(&lhs, ...);
      auto rhs_val = expr->Evaluate(&rhs, ...);

      if (lhs_val.CompareLessThan(rhs_val) == CmpBool::CmpTrue) {
        return order_by_type != OrderByType::DESC;
        // ASC: lhs < rhs, lhs 更好，返回 false（lhs 不应被 pop）
        // 但 priority_queue 是 max-heap，比较器返回 true 意味着 lhs 优先级更低
      }
      // ...
    }
    return true;
  }
};

void TopNExecutor::Init() {
  // 使用反序比较器的最大堆
  auto top_entries = std::priority_queue<Tuple, std::vector<Tuple>, Compare>(Compare(plan_));

  while (child_executor_->Next(&tuple, &rid)) {
    top_entries.push(tuple);
    if (top_entries.size() > plan_->GetN()) {
      top_entries.pop();  // 弹出最差的（最大的），保留最优的 N 个
    }
  }

  // 堆中的元素是逆序的（最差的先出来），需要翻转
  while (!top_entries.empty()) {
    top_tuples_.emplace_back(top_entries.top());
    top_entries.pop();
  }
  std::reverse(top_tuples_.begin(), top_tuples_.end());  // 还原为正确顺序
  curr_pos_ = 0;
}

-- 普通聚合：3 个部门 → 3 行结果
SELECT dept_id, SUM(salary) FROM employees GROUP BY dept_id;

-- 窗口函数：每行都保留，额外计算部门内累计薪资
SELECT emp_id, dept_id, salary,
       SUM(salary) OVER (PARTITION BY dept_id ORDER BY hire_date) AS running_total
FROM employees;
-- 输出与输入等行数，但每行多了 running_total 列

columns_: [col0, col1, placeholder(-1), placeholder(-1)]
           ↑        ↑        ↑                  ↑
       普通列    普通列   窗口函数列1          窗口函数列2

window_functions_: {
  2: WindowFunction{
    function_: SUM(salary)
    type_: SumAggregate
    partition_by_: [dept_id]
    order_by_: [(ASC, hire_date)]
  },
  3: WindowFunction{
    function_: RANK()
    type_: Rank
    partition_by_: [dept_id]
    order_by_: [(DESC, salary)]
  }
}

Key = PARTITION BY 列的值组合（分区键）
Value = 该分区内的累积聚合值

例：PARTITION BY dept_id, ORDER BY hire_date 时：
  key=(10) → value=running_sum_for_dept_10
  key=(20) → value=running_sum_for_dept_20

void WindowFunctionExecutor::Init() {
  // Step 1: 读取所有 child tuples
  while (child_executor_->Next(&tuple, &rid)) {
    tuples.emplace_back(tuple);
  }

  // Step 2: 若有 ORDER BY，对所有 tuples 排序（所有窗口函数共享同一排序）
  if (有 order_by) {
    std::sort(tuples.begin(), tuples.end(), 按 order_by 排序);
  }

  // Step 3: 为每个窗口函数创建 HashTable
  for (each window_function):
    SimpleWindowFunctionHashTable aht = ...;
    if (!has_order_by):
      for (each tuple): aht.InsertCombine(tuple);  // 预聚合
    ahts.push_back(aht);

  // Step 4: 遍历每条 tuple，组装最终输出
  for (each tuple):
    values = []
    for (each output_col):
      if (是普通列): values.push(tuple.GetValue(col_idx))
      else:  // 窗口函数列
        if (has_order_by):
          values.push(aht.InsertCombine(tuple))  // 流式累积
        else:
          values.push(aht.Find(tuple))           // 查预计算结果
    results_.push_back(Tuple(values, output_schema))
}

// SimpleWindowFunctionHashTable::CombineWindowAggregateValues
case WindowFunctionType::Rank:
  ++rank_count_;  // 全局计数器，统计已处理多少行
  if (result->GetAs<int32_t>() != input.GetAs<int32_t>()) {
    // 值变了（不同排名），更新 last_rank_count_ 为当前行号
    *result = input;
    last_rank_count_ = rank_count_;
  }
  // 返回 last_rank_count_（相同值的行返回相同排名）
  return ValueFactory::GetIntegerValue(last_rank_count_);

SimpleWindowFunctionHashTable aht =
    (window_function_type == WindowFunctionType::Rank)
        ? SimpleWindowFunctionHashTable{order_by[0].second, win_type, partition_by, schema}
        : SimpleWindowFunctionHashTable{function, win_type, partition_by, schema};

auto ProjectionExecutor::Next(Tuple *tuple, RID *rid) -> bool {
  Tuple child_tuple{};
  if (!child_executor_->Next(&child_tuple, rid)) { return false; }

  std::vector<Value> values{};
  values.reserve(GetOutputSchema().GetColumnCount());
  for (const auto &expr : plan_->GetExpressions()) {
    // 每个表达式对子 tuple 求值，可以是列引用、运算、常量等
    values.push_back(expr->Evaluate(&child_tuple, child_executor_->GetOutputSchema()));
  }
  *tuple = Tuple{values, &GetOutputSchema()};
  return true;
}

SELECT id, salary * 1.1 AS new_salary FROM employees;
-- GetExpressions() = [ColumnValueExpr(col=0), ArithmeticExpr(ColumnValueExpr(col=2), *, Constant(1.1))]

auto FilterExecutor::Next(Tuple *tuple, RID *rid) -> bool {
  auto filter_expr = plan_->GetPredicate();
  while (true) {
    if (!child_executor_->Next(tuple, rid)) { return false; }
    auto value = filter_expr->Evaluate(tuple, child_executor_->GetOutputSchema());
    if (!value.IsNull() && value.GetAs<bool>()) {
      return true;  // 满足谓词，向上返回
    }
    // 不满足，继续拉取子算子
  }
}

auto ValuesExecutor::Next(Tuple *tuple, RID *rid) -> bool {
  if (cursor_ >= plan_->GetValues().size()) { return false; }

  const auto &row_expr = plan_->GetValues()[cursor_];
  std::vector<Value> values{};
  for (const auto &col : row_expr) {
    // dummy_schema_ 是空 Schema（VALUES 的列不来自任何表）
    values.push_back(col->Evaluate(nullptr, dummy_schema_));
  }
  *tuple = Tuple{values, &GetOutputSchema()};
  cursor_ += 1;
  return true;
}

TableHeap 中的最新版本
  TupleMeta.ts_ = 提交事务的 commit_ts，或未提交事务的 TXN_TEMP_TS
  TupleMeta.is_deleted_ = bool（是否为删除标记）

    ↓ TransactionManager::GetVersionLink(rid)

VersionUndoLink（版本链头）
  prev_ = UndoLink（指向第一个 UndoLog）
  in_progress_ = bool（是否有事务正在修改）

    ↓ UndoLink.prev_log_idx_ / txn_id

UndoLog 1（最近一次修改的逆操作）
  is_deleted_: bool          // 修改前是否已删除
  modified_fields_: vector<bool>  // 哪些列被修改了
  tuple_: Tuple              // 修改前的值（只含 modified 的列）
  ts_: timestamp_t           // 修改前的时间戳
  prev_version_: UndoLink    // 指向更早的 UndoLog

    ↓ prev_version_

UndoLog 2 → UndoLog 3 → ... → nullptr

对于事务 T（read_ts = T.GetReadTs()）读取 tuple 时：

if tuple_ts == T.GetTransactionId():
  → 这是 T 自己修改的，对 T 可见（读己之所写）

elif tuple_ts <= read_ts:
  → 已提交且在 T 开始前提交，对 T 可见

else (tuple_ts > read_ts):
  → 在 T 开始后修改（未提交或其他事务提交）
  → 需要回溯 undo log，找到 ts_ <= read_ts 的版本

if (tuple_ts > txn_ts && tuple_ts != txn->GetTransactionId()) {
  std::vector<UndoLog> undo_logs;
  UndoLink undo_link = txn_manager->GetUndoLink(*rid).value_or(UndoLink{});

  while (optional_undo_log.has_value() && tuple_ts > txn_ts) {
    undo_logs.push_back(*optional_undo_log);
    tuple_ts = optional_undo_log->ts_;         // 更新为该 undo log 之前的时间戳
    undo_link = optional_undo_log->prev_version_;
    optional_undo_log = txn_manager->GetUndoLogOptional(undo_link);
  }

  if (tuple_ts > txn_ts) { continue; }  // 版本链全都太新，不可见

  auto new_tuple = ReconstructTuple(&schema, *tuple, tuple_meta, undo_logs);
  if (!new_tuple.has_value()) { continue; }  // 历史版本是已删除状态
}

auto ReconstructTuple(const Schema *schema, const Tuple &base_tuple,
                      const TupleMeta &base_meta,
                      const std::vector<UndoLog> &undo_logs) -> std::optional<Tuple> {
  bool is_deleted = base_meta.is_deleted_;
  std::vector<Value> values;
  // 初始化为最新版本的所有列值
  for (uint32_t i = 0; i < schema->GetColumnCount(); ++i) {
    values.push_back(base_tuple.GetValue(schema, i));
  }

  for (const auto &undo_log : undo_logs) {
    if (undo_log.is_deleted_) {
      // 这个 undo log 表示在它之前的版本是"已删除"状态
      is_deleted = true;
      for (auto &val : values) val = NULL;
    } else {
      is_deleted = false;
      // 按 modified_fields_ 位图还原被修改的列
      std::vector<uint32_t> cols;
      for (uint32_t i = 0; i < undo_log.modified_fields_.size(); ++i) {
        if (undo_log.modified_fields_[i]) cols.push_back(i);
      }
      Schema undo_schema = Schema::CopySchema(schema, cols);
      for (uint32_t i = 0, j = 0; i < modified_fields.size(); ++i) {
        if (modified_fields[i]) {
          values[i] = undo_log.tuple_.GetValue(&undo_schema, j++);
        }
      }
    }
  }

  if (is_deleted) return std::nullopt;
  Tuple res = Tuple(values, schema);
  res.SetRid(base_tuple.GetRid());
  return res;
}

// 若 tuple_ts > txn->GetReadTs() 且不是本事务修改的
// 说明在 T 的 snapshot 之后，有另一个事务（已提交或未提交）修改了这条 tuple
if (tuple_meta.ts_ > txn->GetReadTs() && tuple_meta.ts_ != txn->GetTransactionId()) {
  txn->SetTainted();
  throw ExecutionException("write-write conflict");
}

1. Filter Predicate Pushdown     → 将 Filter 下推至 SeqScan
2. Column Pruning                → 删除不必要的列（可选）
3. OptimizeSeqScanAsIndexScan    → SeqScan+等值谓词 → IndexScan
4. OptimizeNLJAsHashJoin         → NestedLoopJoin+等值条件 → HashJoin
5. OptimizeSortLimitAsTopN       → Sort+Limit → TopN
6. ...

auto Optimizer::OptimizeSeqScanAsIndexScan(const AbstractPlanNodeRef &plan)
    -> AbstractPlanNodeRef {
  // 递归优化子节点
  std::vector<AbstractPlanNodeRef> children;
  for (const auto &child : plan->GetChildren()) {
    children.emplace_back(OptimizeSeqScanAsIndexScan(child));
  }
  auto optimized_plan = plan->CloneWithChildren(children);

  // 匹配：SeqScan + 谓词为等值比较（col = const）
  if (optimized_plan->GetType() == PlanType::SeqScan) {
    const auto &seq_scan_plan = dynamic_cast<SeqScanPlanNode &>(*optimized_plan);
    auto predicate = seq_scan_plan.filter_predicate_;

    if (predicate != nullptr) {
      auto comparison_expr = std::dynamic_pointer_cast<ComparisonExpression>(predicate);
      if (comparison_expr != nullptr && comparison_expr->comp_type_ == ComparisonType::Equal) {
        auto column_expr = std::dynamic_pointer_cast<ColumnValueExpression>(comparison_expr->GetChildAt(0));
        auto constant_expr = std::dynamic_pointer_cast<ConstantValueExpression>(comparison_expr->GetChildAt(1));

        if (column_expr != nullptr && constant_expr != nullptr) {
          auto column_idx = column_expr->GetColIdx();
          // 查找该列上的索引
          for (auto &index_info : catalog_.GetTableIndexes(table_name)) {
            auto attrs = index_info->index_->GetKeyAttrs();
            if (std::find(attrs.begin(), attrs.end(), column_idx) != attrs.end()) {
              // 创建 IndexScanPlanNode 替换 SeqScanPlanNode
              return std::make_shared<IndexScanPlanNode>(
                  output_schema, table_oid, index_oid, predicate, constant_expr.get());
            }
          }
        }
      }
    }
  }
  return optimized_plan;
}

NLJ(condition: a1=b1 AND a2=b2)
→ HashJoin(left_keys=[a1, a2], right_keys=[b1, b2])

Limit(n=10)
  └─ Sort(ORDER BY salary DESC)

→ TopN(n=10, ORDER BY salary DESC)