Untitled

#include "execution/executors/topn_executor.h"

namespace bustub {

TopNExecutor::TopNExecutor(ExecutorContext *exec_ctx, const TopNPlanNode *plan,
                           std::unique_ptr<AbstractExecutor> &&child_executor)
    : AbstractExecutor(exec_ctx) {
  plan_ = plan;
  child_executor_ = std::move(child_executor);
  first_call_ = false;
  count_ = 0;
}

void TopNExecutor::Init() {
  first_call_ = true;
  child_executor_->Init();

  Tuple tuple;
  RID rid;
  while (child_executor_->Next(&tuple, &rid)) {
    tuples_.emplace_back(tuple);
  }

  std::sort(tuples_.begin(), tuples_.end(), [this](const Tuple &a, const Tuple &b) {
    const auto &order_by_list = plan_->GetOrderBy();

    for (const auto &[order_by_type, expr] : order_by_list) {
      Value val_a = expr->Evaluate(&a, plan_->OutputSchema());
      Value val_b = expr->Evaluate(&b, plan_->OutputSchema());

      if (order_by_type == OrderByType::DEFAULT || order_by_type == OrderByType::ASC) {
        if (val_a.CompareLessThan(val_b) == CmpBool::CmpTrue) {
          return true;
        }
        if (val_a.CompareGreaterThan(val_b) == CmpBool::CmpTrue) {
          return false;
        }
      } else {
        if (val_a.CompareLessThan(val_b) == CmpBool::CmpTrue) {
          return false;
        }
        if (val_a.CompareGreaterThan(val_b) == CmpBool::CmpTrue) {
          return true;
        }
      }

      // If the current columns are equal, move to the next column
    }
    return false;
  });

  current_tuple_idx_ = 0;
  count_ = 0;
}

auto TopNExecutor::Next(Tuple *tuple, RID *rid) -> bool {
  if (current_tuple_idx_ >= static_cast<int>(tuples_.size())) {
    return false;
  }

  if (count_ < plan_->GetN()) {
    *tuple = tuples_[current_tuple_idx_];
    *rid = tuple->GetRid();
    count_++;
    current_tuple_idx_++;
    return true;
  }

  return false;
}

auto TopNExecutor::GetNumInHeap() -> size_t {  // ?????
  return count_;
};

}  // namespace bustub