/**
 * Slotted page format:
 *  ---------------------------------------------------------
 *  | HEADER | ... FREE SPACE ... | ... INSERTED TUPLES ... |
 *  ---------------------------------------------------------
 *                                ^
 *                                free space pointer
 *
 *  Header format (size in bytes):
 *  ----------------------------------------------------------------------------
 *  | NextPageId (4)| NumTuples(2) | NumDeletedTuples(2) |
 *  ----------------------------------------------------------------------------
 *  ----------------------------------------------------------------
 *  | Tuple_1 offset+size (4) | Tuple_2 offset+size (4) | ... |
 *  ----------------------------------------------------------------
 *
 * Tuple format:
 * | meta | data |
 */

class LRUKNode {
 private:
  /** History of last seen K timestamps of this page. Least recent timestamp stored in front. */
  std::list<size_t> history_;         // history timestamp
  size_t k_{0};                       // k times
  frame_id_t fid_{INVALID_FRAME_ID};  // frame id
  bool is_evictable_{false};          // whether or not can be evicted
};

class LRUKReplacer {
  std::unordered_map<frame_id_t, LRUKNode> node_store_;
  size_t current_timestamp_{0};
  size_t curr_size_{0};
  size_t replacer_size_;
  size_t k_;
  std::mutex latch_;
};

struct DiskRequest {
  /** Flag indicating whether the request is a write or a read. */
  bool is_write_;
  /**
   *  Pointer to the start of the memory location where a page is either:
   *   1. being read into from disk (on a read).
   *   2. being written out to disk (on a write).
   */
  char *data_;
  /** ID of the page being read from / written to disk. */
  page_id_t page_id_;
  /** Callback used to signal to the request issuer when the request has been completed. */
  std::promise<bool> callback_;
};

void DiskScheduler::StartWorkerThread() {
  std::optional<DiskRequest> request;
  while (true) {
    request = request_queue_.Get();
    if (request != std::nullopt) {
      if (request->is_write_) {
        disk_manager_->WritePage(request->page_id_, request->data_);
        request->callback_.set_value(true);
      } else if (!request->is_write_) {
        disk_manager_->ReadPage(request->page_id_, request->data_);
        request->callback_.set_value(true);
      }
    } else {
      break;
    }
  }
}

/** @brief A set of dirty pages that need to be flushed to disk. */
std::unordered_set<page_id_t> dirty_pages_;
/** @brief A mutex to protect the dirty pages set. */
std::mutex flush_mutex_;
/** @brief A condition variable to notify the flusher thread that there are dirty pages to flush. */
std::condition_variable flush_cv_;

INDEX_TEMPLATE_ARGUMENTS
auto BPLUSTREE_TYPE::OptiInsert(Context *ctx, const KeyType &key, const ValueType &value) -> bool {
  ctx->header_page_ = bpm_->WritePage(header_page_id_);
  if (IsEmpty(ctx)) {
    return StartNewTree(ctx, key, value); // 如果是空树，直接开始新树
  }
  ctx->ReleaseAllLatch();
  auto leaf_page_id = OptiFindLeafPage(ctx, Operation::INSERT, key, comparator_, false, false); // 读锁
  // 此时ctx.write_set_中保存这leaf_page的写锁
  auto leaf_page = ctx->write_set_.back().AsMut<LeafPage>();
  if (leaf_page->GetSize() < leaf_page->GetMaxSize() - 1) {
    auto before_insert_size = leaf_page->GetSize();
    leaf_page->Insert(key, value, comparator_);
    auto new_size = leaf_page->GetSize();
    // check leaf node whether it is full or not
    // 1. duplicate key
    return before_insert_size != new_size;
  }
  ctx->ReleaseAllLatch();
  ctx->header_page_ = bpm_->WritePage(header_page_id_);

  auto new_leaf_page_id = FindLeafPage(ctx, Operation::INSERT, key, comparator_, false, false); // 写锁

  // 如果找到叶子节点，一定确保ctx.write_set_不为空
  BUSTUB_ASSERT(!ctx->write_set_.empty(), "write set is empty");
  auto new_leaf_page = ctx->write_set_.back().template AsMut<LeafPage>();

  auto before_insert_size = new_leaf_page->GetSize();
  new_leaf_page->Insert(key, value, comparator_);
  auto new_size = new_leaf_page->GetSize();
  // check leaf node whether it is full or not
  // 1. duplicate key
  if (before_insert_size == new_size) {
    return false;
  }
  // 2. leaf node is  not full
  if (new_size < new_leaf_page->GetMaxSize()) {
    return true;
  }
  // 3. leaf node is full
  auto ret = SplitLeaf(ctx);
  auto new_key = ret.new_key_;
  auto new_page_id = ret.new_page_id_;

  InsertIntoParent(ctx, new_leaf_page_id, new_key, new_page_id, 0);
  return true;
}

INDEX_TEMPLATE_ARGUMENTS
void BPLUSTREE_TYPE::OptiRemove(Context *ctx, const KeyType &key) {
  ctx->header_page_ = bpm_->WritePage(header_page_id_);
  if (IsEmpty(ctx)) {
    return;
  }
  ctx->ReleaseAllLatch();
  auto leaf_page_id = OptiFindLeafPage(ctx, Operation::REMOVE, key, comparator_, false, false);

  auto leaf_page = ctx->write_set_.back().AsMut<LeafPage>();
  if (leaf_page->GetSize() > leaf_page->GetMinSize()) {
    leaf_page->RemoveAndDeleteRecord(key, comparator_);
    return;
  }
  ctx->ReleaseAllLatch();
  ctx->header_page_ = bpm_->WritePage(header_page_id_);
  auto new_leaf_page_id = FindLeafPage(ctx, Operation::REMOVE, key, comparator_, false, false);
  if (leaf_page_id == INVALID_PAGE_ID) {
    return;
  }

  DeleteEntry(ctx, key, -1, new_leaf_page_id, 0);
}

INDEX_TEMPLATE_ARGUMENTS
auto BPLUSTREE_TYPE::DeleteEntry(Context *ctx, const KeyType &key, int delete_index, page_id_t current_page_id,
                                 int recursive_level) -> bool {
  if (ctx->IsRootPage(current_page_id)) {
    AdjustRoot(ctx);
    return true;
  }
  // ...
  bool redistribute = Redistribute(ctx, old_page_index, sibling_index, recursive_level, parent_key);
  if (!redistribute) {
    if (old_page_index != 0) {
      coalesec = Coalesce(ctx, sibling_index, true, recursive_level, parent_key);
    } else {
      coalesec = Coalesce(ctx, sibling_index, false, recursive_level, parent_key);
    }
  }
  return true;
}

INDEX_TEMPLATE_ARGUMENTS
class IndexIterator {
 private:
  // add your own private member variables here
  BufferPoolManager *bpm_;
  std::optional<ReadPageGuard> leaf_page_guard_;
  int index_;
};

INDEX_TEMPLATE_ARGUMENTS
auto BPLUSTREE_TYPE::Begin(const KeyType &key) -> INDEXITERATOR_TYPE {
  Context ctx;
  FindLeafPage(&ctx, Operation::SEARCH, key, comparator_, false, false);
  auto *leaf_page = ctx.read_set_.back().As<LeafPage>();
  auto index = leaf_page->KeyIndex(key, comparator_);
  return INDEXITERATOR_TYPE(bpm_, std::optional<ReadPageGuard>(std::move(ctx.read_set_.back())), index);
}

INDEX_TEMPLATE_ARGUMENTS
auto BPLUSTREE_TYPE::End() -> INDEXITERATOR_TYPE {
  return INDEXITERATOR_TYPE(bpm_, std::nullopt, 0);
}

/**
 * Slotted page format:
 *  ---------------------------------------------------------
 *  | HEADER | ... FREE SPACE ... | ... INSERTED TUPLES ... |
 *  ---------------------------------------------------------
 *                                ^
 *                                free space pointer
 *
 *  Header format (size in bytes):
 *  ----------------------------------------------------------------------------
 *  | NextPageId (4)| NumTuples(2) | NumDeletedTuples(2) |
 *  ----------------------------------------------------------------------------
 *  ----------------------------------------------------------------
 *  | Tuple_1 offset+size (4) | Tuple_2 offset+size (4) | ... |
 *  ----------------------------------------------------------------
 *
 * Tuple format:
 * | meta | data |
 */
 struct TupleMeta {
  /** the ts / txn_id of this tuple. In project 3, simply set it to 0. */
  timestamp_t ts_;
  /** marks whether this tuple is marked removed from table heap. */
  bool is_deleted_;
 };

 class Tuple {
  RID rid_{};  // if pointing to the table heap, the rid is valid
  std::vector<char> data_;
};

void HashJoinExecutor::Init() {
  // 1. init child executors
  left_child_executor_->Init();
  right_child_executor_->Init();

  // 2. build hash table for right table
  Tuple right_tuple{};
  RID right_rid;
  while (right_child_executor_->Next(&right_tuple, &right_rid)) {
    HashJoinKey key = GetRightJoinKey(&right_tuple);
    // jht_ 是 HashJoinTable 的实例，存储了右表的哈希表
    jht_->Insert(key, right_tuple);
  }
}

auto HashJoinExecutor::Next(Tuple *tuple, RID *rid) -> bool {
  if (!results_.empty()) {
    *tuple = results_.front();
    *rid = RID{1, 0};
    results_.pop_front();
    return true;
  }
  auto left_schema = left_child_executor_->GetOutputSchema();
  auto right_schema = right_child_executor_->GetOutputSchema();
  while (left_child_executor_->Next(tuple, rid)) {
    auto left_hash_key = GetLeftJoinKey(tuple);
    if (jht_->GetValue(left_hash_key) != nullptr) {
      for (const auto &right_tuple : *jht_->GetValue(left_hash_key)) {
        std::vector<Value> values;
        for (uint32_t i = 0; i < left_schema.GetColumnCount(); i++) {
          values.emplace_back(tuple->GetValue(&left_schema, i));
        }
        for (uint32_t i = 0; i < right_schema.GetColumnCount(); i++) {
          values.emplace_back(right_tuple.GetValue(&right_schema, i));
        }
        results_.emplace_back(Tuple{values, &GetOutputSchema()});
      }
    } else if (plan_->GetJoinType() == JoinType::LEFT) {
      std::vector<Value> values;
      for (uint32_t i = 0; i < left_schema.GetColumnCount(); i++) {
        values.emplace_back(tuple->GetValue(&left_schema, i));
      }
      for (uint32_t i = 0; i < right_schema.GetColumnCount(); i++) {
        values.emplace_back(ValueFactory::GetNullValueByType(right_schema.GetColumn(i).GetType()));
      }
      results_.emplace_back(Tuple{values, &GetOutputSchema()});
    }
    // 直到这次的left_tuple与right_tuples匹配的数组都被遍历完才会继续下一次的left_tuple
    if (!results_.empty()) {
      *tuple = results_.front();
      *rid = RID{1, 0};
      results_.pop_front();
      return true;
    }
  }
  return false;
}

template <size_t K>
void ExternalMergeSortExecutor<K>::Init() {
  child_executor_->Init();
  // 1. 生成初始的mergesortrun
  auto *bpm = exec_ctx_->GetBufferPoolManager();
  // ...
  while (child_executor_->Next(&child_tuple, &child_rid)) {
    auto sort_key = GenerateSortKey(child_tuple, plan_->GetOrderBy(), GetOutputSchema());
    buffer.emplace_back(SortEntry{sort_key, child_tuple});
    if (buffer.size() == max_tuples_per_page) {
      task_nums++;
      // 将buffer中的数据flush到一个新的MergeSortRun中
      thread_pool_->Enqueue([=]() {
        FlushBufferToRun(buffer, bpm, max_tuples_per_page, tuple_length, runs_);
        task_nums_.fetch_add(1);
      });
      buffer.clear();
    }
  }
  // 未满一页也需要flush
  if (!buffer.empty()) {
    task_nums++;
    thread_pool_->Enqueue([=]() {
      FlushBufferToRun(buffer, bpm, max_tuples_per_page, tuple_length, runs_);
      task_nums_.fetch_add(1);
    });
    buffer.clear();
  }
  // 同步IO线程，保证所有的page都是排好序的
  while (task_nums != task_nums_.load()) {
  }
  task_nums_.store(0);

  // 2. merge all runs
  while (runs_.size() > 1) {
    std::vector<MergeSortRun> new_runs;
    // 2-way external merge sort
    for (size_t i = 0; i < runs_.size(); i += 2) {
      if (i + 1 < runs_.size()) {
        Merge2Runs(runs_[i], runs_[i + 1], bpm, max_tuples_per_page, tuple_length, new_runs);
      } else {
        new_runs.emplace_back(runs_[i]);
      }
    }
    runs_ = std::move(new_runs);
  }

  // 3. read the last run
  if (!runs_.empty()) {
    current_iterator_ = runs_[0].Begin();
    end_iterator_ = runs_[0].End();
    is_inited_ = true;
  } else {
    current_iterator_ = {};
    end_iterator_ = {};
  }
}

/**
 * TransactionManager keeps track of all the transactions running in the system.
 */
class TransactionManager {
 public:
  /**
   * Begins a new transaction.
   * @param isolation_level an optional isolation level of the transaction.
   * @return an initialized transaction
   */
  auto Begin(IsolationLevel isolation_level = IsolationLevel::SNAPSHOT_ISOLATION) -> Transaction *;

  /**
   * Commits a transaction.
   * @param txn the transaction to commit, the txn will be managed by the txn manager so no need to delete it by
   * yourself
   */
  auto Commit(Transaction *txn) -> bool;

  /**
   * Aborts a transaction
   * @param txn the transaction to abort, the txn will be managed by the txn manager so no need to delete it by yourself
   */
  void Abort(Transaction *txn);

  /** @brief Stop-the-world garbage collection. Will be called only when all transactions are not accessing the table
   * heap. */
  void GarbageCollection();

  /** protects txn map */
  std::shared_mutex txn_map_mutex_;
  /** All transactions, running or committed */
  std::unordered_map<txn_id_t, std::shared_ptr<Transaction>> txn_map_;

  struct PageVersionInfo {
    /** protects the map */
    std::shared_mutex mutex_;
    /** Stores previous version info for all slots. Note: DO NOT use `[x]` to access it because
     * it will create new elements even if it does not exist. Use `find` instead.
     */
    std::unordered_map<slot_offset_t, UndoLink> prev_link_;
  };

  /** protects version info */
  std::shared_mutex version_info_mutex_;
  /** Stores the previous version of each tuple in the table heap. Do not directly access this field. Use the helper
   * functions in `transaction_manager_impl.cpp`. */
  std::unordered_map<page_id_t, std::shared_ptr<PageVersionInfo>> version_info_;

  /** Stores all the read_ts of running txns so as to facilitate garbage collection. */
  Watermark running_txns_{0};

  /** Only one txn is allowed to commit at a time */
  std::mutex commit_mutex_;
  /** The last committed timestamp. */
  std::atomic<timestamp_t> last_commit_ts_{0};

  /** Catalog */
  Catalog *catalog_;

  std::atomic<txn_id_t> next_txn_id_{TXN_START_ID};
};

/**
 * Transaction tracks information related to a transaction.
 */
class Transaction {
  // The below fields should be ONLY changed by txn manager (with the txn manager lock held).

  /** The state of this transaction. */
  std::atomic<TransactionState> state_{TransactionState::RUNNING};

  /** The read ts */
  std::atomic<timestamp_t> read_ts_{0};

  /** The commit ts */
  std::atomic<timestamp_t> commit_ts_{INVALID_TS};

  /** The latch for this transaction for accessing txn-level undo logs, protecting all fields below. */
  std::mutex latch_;

  /**
   * @brief Store undo logs. Other undo logs / table heap will store (txn_id, index) pairs, and therefore
   * you should only append to this vector or update things in-place without removing anything.
   */
  std::vector<UndoLog> undo_logs_;

  /** stores the RID of write tuples */
  std::unordered_map<table_oid_t, std::unordered_set<RID>> write_set_;
  /** store all scan predicates */
  std::unordered_map<table_oid_t, std::vector<AbstractExpressionRef>> scan_predicates_;

  // The below fields are set when a txn is created and will NEVER be changed.

  /** The isolation level of the transaction. */
  const IsolationLevel isolation_level_;

  /** The thread ID which the txn starts from.  */
  const std::thread::id thread_id_;

  /** The ID of this transaction. */
  const txn_id_t txn_id_;
};