diff --git a/deps/common/io/io.cpp b/deps/common/io/io.cpp
index e256c19cab1e2f9c972ec46654f4187fd1f00536..0a531c6c086426b2e10fe1ee0dfbf747f4b8a50b 100644
--- a/deps/common/io/io.cpp
+++ b/deps/common/io/io.cpp
@@ -349,4 +349,40 @@ int getFileSize(const char *filePath, u64_t &fileLen)
   return 0;
 }
 
-}  // namespace common
\ No newline at end of file
+int writen(int fd, const void *buf, int size)
+{
+  const char *tmp = (const char *)buf;
+  while ( size > 0) {
+    const ssize_t ret = ::write(fd, tmp, size);
+    if (ret >= 0) {
+      tmp  += ret;
+      size -= ret;
+      continue;
+    }
+    const int err = errno;
+    if (EAGAIN != err && EINTR != err)
+      return err;
+   }
+  return 0;
+}
+
+int readn(int fd, void *buf, int size)
+{
+  char *tmp = (char *)buf;
+  while (size > 0) {
+    const ssize_t ret = ::read(fd, tmp, size);
+    if (ret > 0) {
+      tmp  += ret;
+      size -= ret;
+      continue;
+    }
+    if (0 == ret)
+      return -1; // end of file
+
+    const int err = errno;
+    if (EAGAIN != err && EINTR != err)
+      return err;
+  }
+  return 0;
+}
+}  // namespace common
diff --git a/deps/common/io/io.h b/deps/common/io/io.h
index e4f01702ddd8c06dee5eef7f41bd735c78569e07..390551c0b1b94b71acf3ea2d190781bc0f2b2aab 100644
--- a/deps/common/io/io.h
+++ b/deps/common/io/io.h
@@ -58,5 +58,8 @@ int touch(const std::string &fileName);
  */
 int getFileSize(const char *filePath, u64_t &fileLen);
 
+int writen(int fd, const void *buf, int size);
+int readn(int fd, void *buf, int size);
+
 }  // namespace common
 #endif /* __COMMON_IO_IO_H__ */
diff --git a/deps/common/lang/lower_bound.h b/deps/common/lang/lower_bound.h
new file mode 100644
index 0000000000000000000000000000000000000000..5ebbe6822c9f0bbad0d32dac15b07276ab2e0b95
--- /dev/null
+++ b/deps/common/lang/lower_bound.h
@@ -0,0 +1,115 @@
+/* Copyright (c) 2021 Xie Meiyi(xiemeiyi@hust.edu.cn) and OceanBase and/or its affiliates. All rights reserved.
+miniob is licensed under Mulan PSL v2.
+You can use this software according to the terms and conditions of the Mulan PSL v2.
+You may obtain a copy of Mulan PSL v2 at:
+         http://license.coscl.org.cn/MulanPSL2
+THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
+EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
+MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
+See the Mulan PSL v2 for more details. */
+
+//
+//
+// Created by Wangyunlai
+//
+
+#pragma once
+
+#include <iterator>
+
+namespace common {
+/**
+ * @brief 找到不小于指定值的元素
+ * @note  数组中的元素都不相等
+ * @param first 已经排序的序列中第一个元素
+ * @param last  已经排序的序列中的最后一个元素的后一个
+ * @param val   要查找的元素的值
+ * @param comp  比较函数。相等返回0，小于返回负数，大于返回正数
+ * @param _found 如果给定，会返回是否存在于当前的序列中
+ * @return ForwardIterator 指向lower bound结果的iterator。如果大于最大的值，那么会指向last
+ */
+template <typename ForwardIterator, typename T, typename Compare>
+ForwardIterator lower_bound(ForwardIterator first, ForwardIterator last,
+			    const T &val, Compare comp, bool *_found = nullptr)
+{
+  bool found = false;
+  ForwardIterator iter;
+  const auto count = std::distance(first, last);
+  auto last_count = count;
+  while (last_count > 0) {
+    iter = first;
+    auto step = last_count / 2;
+    std::advance(iter, step);
+    int result = comp(*iter, val);
+    if (0 == result) {
+      first = iter;
+      found = true;
+      break;
+    }
+    if (result < 0) {
+      first = ++iter;
+      last_count -= step + 1;
+    } else {
+      last_count = step;
+    }
+  }
+
+  if (_found) {
+    *_found = found;
+  }
+  return first;
+}
+
+template <typename T>
+class Comparator
+{
+public:
+  int operator() (const T &v1, const T &v2) const
+  {
+    if (v1 < v2) {
+      return -1;
+    }
+    if (v1 > v2) {
+      return 1;
+    }
+    return 0;
+  }
+};
+
+template <typename ForwardIterator, typename T>
+ForwardIterator lower_bound(ForwardIterator first, ForwardIterator last, const T &val, bool *_found = nullptr)
+{
+  return lower_bound<ForwardIterator, T, Comparator<T>>(first, last, val, Comparator<T>(), _found);
+}
+
+template <typename T, typename Distance = ptrdiff_t>
+class BinaryIterator : public std::iterator<std::random_access_iterator_tag, T *, Distance>
+{
+public: 
+  BinaryIterator() = default;
+  BinaryIterator(size_t item_num, T *data) : item_num_(item_num), data_(data)
+  {}
+
+  BinaryIterator &operator+= (int n) { data_ += (item_num_ * n); return *this; }
+  BinaryIterator &operator-= (int n) { return this->operator+ (-n); }
+  BinaryIterator &operator++() { return this->operator +=(1); }
+  BinaryIterator operator++(int) { BinaryIterator tmp(*this); this->operator++(); return tmp; }
+  BinaryIterator &operator--() { return this->operator += (-1); }
+  BinaryIterator operator--(int) { BinaryIterator tmp(*this); this->operator += (-1); return tmp; }
+
+  bool operator == (const BinaryIterator &other) const { return data_ == other.data_; }
+  bool operator != (const BinaryIterator &other) const { return ! (this->operator ==(other)); }
+
+  T * operator *() { return data_; }
+  T * operator ->() { return data_; }
+
+  friend Distance operator - (const BinaryIterator &left, const BinaryIterator &right)
+  {
+    return (left.data_ - right.data_) / left.item_num_;
+  }
+
+private:
+  size_t item_num_ = 0;
+  T * data_ = nullptr;
+};
+} // namespace common
diff --git a/deps/common/lang/lru_cache.h b/deps/common/lang/lru_cache.h
new file mode 100644
index 0000000000000000000000000000000000000000..a958c05a8082869b143cf24aa172073f05d15de0
--- /dev/null
+++ b/deps/common/lang/lru_cache.h
@@ -0,0 +1,227 @@
+/* Copyright (c) 2021 Xie Meiyi(xiemeiyi@hust.edu.cn) and OceanBase and/or its affiliates. All rights reserved.
+miniob is licensed under Mulan PSL v2.
+You can use this software according to the terms and conditions of the Mulan PSL v2.
+You may obtain a copy of Mulan PSL v2 at:
+         http://license.coscl.org.cn/MulanPSL2
+THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
+EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
+MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
+See the Mulan PSL v2 for more details. */
+
+//
+// Created by Wangyunlai on 2022.09.02
+//
+
+#pragma once
+
+#include <functional>
+#include <unordered_set>
+
+namespace common {
+
+template <typename Key, typename Value,
+          typename Hash = std::hash<Key>,
+	  typename Pred = std::equal_to<Key>>
+class LruCache {
+
+  class ListNode {
+  public:
+    Key   key_;
+    Value value_;
+
+    ListNode *prev_ = nullptr;
+    ListNode *next_ = nullptr;
+
+  public:
+    ListNode(const Key &key, const Value &value) : key_(key), value_(value)
+    {}
+  };
+
+  class PListNodeHasher {
+  public:
+    size_t operator() (ListNode *node) const {
+      if (node == nullptr) {
+	return 0;
+      }
+      return hasher_(node->key_);
+    }
+
+  private:
+    Hash hasher_;
+  };
+
+  class PListNodePredicator {
+  public:
+    bool operator() (ListNode * const node1, ListNode * const node2) const {
+      if (node1 == node2) {
+	return true;
+      }
+
+      if (node1 == nullptr || node2 == nullptr) {
+	return false;
+      }
+
+      return pred_(node1->key_, node2->key_);
+    }
+  private:
+    Pred pred_;
+  };
+
+public:  
+  LruCache(size_t reserve = 0)
+  {
+    if (reserve > 0) {
+      searcher_.reserve(reserve);   
+    }
+  }
+
+  ~LruCache()
+  {
+    destroy();
+  }
+
+  void destroy()
+  {
+    for (ListNode *node : searcher_) {
+      delete node;
+    }
+    searcher_.clear();
+
+    lru_front_ = nullptr;
+    lru_tail_  = nullptr;
+  }
+
+  size_t count() const
+  {
+    return searcher_.size();
+  }
+
+  bool get(const Key &key, Value &value)
+  {
+    auto iter = searcher_.find((ListNode *)&key);
+    if (iter == searcher_.end()) {
+      return false;
+    }
+
+    lru_touch(*iter);
+    value = (*iter)->value_;
+    return true;
+  }
+
+  void put(const Key &key, const Value &value)
+  {
+    auto iter = searcher_.find((ListNode *)&key);
+    if (iter != searcher_.end()) {
+      ListNode * ln = *iter;
+      ln->value_ = value;
+      lru_touch(ln);
+      return ;
+    }
+
+    ListNode *ln = new ListNode(key, value);
+    lru_push(ln);
+  }
+
+  void remove(const Key &key)
+  {
+    auto iter = searcher_.find((ListNode *)&key);
+    if (iter != searcher_.end()) {
+      lru_remove(*iter);
+    }
+  }
+
+  void pop(Value *&value)
+  {
+    // TODO
+    value = nullptr;
+  }
+
+  void foreach(std::function<bool(const Key &, const Value &)> func)
+  {
+    for (ListNode *node = lru_front_; node != nullptr; node = node->next_) {
+      bool ret = func(node->key_, node->value_);
+      if (!ret) {
+        break;
+      }
+    }
+  }
+
+  void foreach_reverse(std::function<bool(const Key &, const Value &)> func)
+  {
+    for (ListNode *node = lru_tail_; node != nullptr; node = node->prev_) {
+      bool ret = func(node->key_, node->value_);
+      if (!ret) {
+        break;
+      }
+    }
+  }
+
+private:
+  void lru_touch(ListNode *node)
+  {
+    // move node to front
+    if (nullptr == node->prev_) {
+      return;
+    }
+
+    node->prev_->next_ = node->next_;
+    
+    if (node->next_ != nullptr) {
+      node->next_->prev_ = node->prev_;
+    } else {
+      lru_tail_ = node->prev_;
+    }
+
+    node->prev_ = nullptr;
+    node->next_ = lru_front_;
+    if (lru_front_ != nullptr) {
+      lru_front_->prev_ = node;
+    }
+    lru_front_ = node;
+  }
+
+  void lru_push(ListNode *node)
+  {
+    // push front
+    if (nullptr == lru_tail_) {
+      lru_tail_ = node;
+    }
+
+    node->prev_ = nullptr;
+    node->next_ = lru_front_;
+    if (lru_front_ != nullptr) {
+      lru_front_->prev_ = node;
+    }
+
+    lru_front_ = node;
+    searcher_.insert(node);
+  }
+
+  void lru_remove(ListNode *node)
+  {
+    if (node->prev_ != nullptr) {
+      node->prev_->next_ = node->next_;
+    }
+
+    if (node->next_ != nullptr) {
+      node->next_->prev_ = node->prev_;
+    }
+
+    if (lru_front_ == node) {
+      lru_front_ = node->next_;
+    }
+    if (lru_tail_ == node) {
+      lru_tail_ = node->prev_;
+    }
+
+    searcher_.erase(node);
+  }
+
+private:
+  using SearchType = std::unordered_set<ListNode *, PListNodeHasher, PListNodePredicator>;
+  SearchType   searcher_;
+  ListNode   * lru_front_ = nullptr;
+  ListNode   * lru_tail_  = nullptr;
+};
+
+} // namespace common
diff --git a/deps/common/mm/mem_pool.h b/deps/common/mm/mem_pool.h
index 7b8b6f95557e7614295f9e4923753fbba5045c2d..eadc06597e4e8791b4b10dd14a6c1a8e326a58a5 100644
--- a/deps/common/mm/mem_pool.h
+++ b/deps/common/mm/mem_pool.h
@@ -82,22 +82,6 @@ public:
    */
   virtual void free(T *item) = 0;
 
-  /**
-   * Find first used item which match the function of "func"
-   * @return
-   */
-  virtual T *find(match func, void *arg) = 0;
-
-  /**
-   * Find used item which match the function of "func"
-   */
-  virtual std::list<T *> find_all(match func, void *arg) = 0;
-
-  /**
-   * Mark the item has been changed;
-   */
-  virtual void mark_modified(T *item) = 0;
-
   /**
    * Print the MemPool status
    * @return
@@ -142,7 +126,7 @@ public:
    * init memory pool, the major job is to alloc memory for memory pool
    * @param pool_num, memory pool's number
    * @param item_num_per_pool, how many items per pool.
-   * @return
+   * @return 0 for success and others failure
    */
   int init(bool dynamic = true, int pool_num = DEFAULT_POOL_NUM, int item_num_per_pool = DEFAULT_ITEM_NUM_PER_POOL);
 
@@ -168,22 +152,6 @@ public:
    */
   void free(T *item);
 
-  /**
-   * Find first used item which match the function of "func"
-   * @return
-   */
-  T *find(match func, void *arg);
-
-  /**
-   * Find used item which match the function of "func"
-   */
-  std::list<T *> find_all(match func, void *arg);
-
-  /**
-   * Mark the item has been changed;
-   */
-  void mark_modified(T *item);
-
   /**
    * Print the MemPool status
    * @return
@@ -205,7 +173,6 @@ public:
 
 protected:
   std::list<T *> pools;
-  std::list<T *> lru_used;
   std::set<T *> used;
   std::list<T *> frees;
   int item_num_per_pool;
@@ -255,7 +222,6 @@ void MemPoolSimple<T>::cleanup()
 
   MUTEX_LOCK(&this->mutex);
 
-  lru_used.clear();
   used.clear();
   frees.clear();
   this->size = 0;
@@ -322,7 +288,6 @@ T *MemPoolSimple<T>::alloc()
   frees.pop_front();
 
   used.insert(buffer);
-  lru_used.push_back(buffer);
 
   MUTEX_UNLOCK(&this->mutex);
   return buffer;
@@ -341,69 +306,12 @@ void MemPoolSimple<T>::free(T *buf)
     return;
   }
 
-  lru_used.remove(buf);
   frees.push_back(buf);
 
   MUTEX_UNLOCK(&this->mutex);
   return;  // TODO for test
 }
 
-template <class T>
-T *MemPoolSimple<T>::find(match func, void *arg)
-{
-  T *buffer = nullptr;
-  MUTEX_LOCK(&this->mutex);
-  for (typename std::list<T *>::iterator it = lru_used.begin(); it != lru_used.end(); ++it) {
-    T *item = *it;
-
-    if ((*func)(item, arg) == false) {
-      continue;
-    }
-
-    buffer = item;
-    break;
-  }
-  MUTEX_UNLOCK(&this->mutex);
-  return buffer;
-}
-
-template <class T>
-std::list<T *> MemPoolSimple<T>::find_all(match func, void *arg)
-
-{
-  std::list<T *> ret;
-  MUTEX_LOCK(&this->mutex);
-  for (typename std::list<T *>::iterator it = lru_used.begin(); it != lru_used.end(); ++it) {
-    T *item = *it;
-
-    if ((*func)(item, arg) == false) {
-      continue;
-    }
-
-    ret.push_back(item);
-  }
-  MUTEX_UNLOCK(&this->mutex);
-  return ret;
-}
-
-template <class T>
-void MemPoolSimple<T>::mark_modified(T *buf)
-{
-  MUTEX_LOCK(&this->mutex);
-
-  if (used.find(buf) == used.end()) {
-    MUTEX_UNLOCK(&this->mutex);
-    LOG_WARN("No entry of %p in %s.", buf, this->name.c_str());
-    return;
-  }
-
-  lru_used.remove(buf);
-  lru_used.push_back(buf);
-
-  MUTEX_UNLOCK(&this->mutex);
-  return;
-}
-
 template <class T>
 std::string MemPoolSimple<T>::to_string()
 {
diff --git a/src/observer/net/server.cpp b/src/observer/net/server.cpp
index 2c6cef6dd9bb452b43e5d0125f30851a8fcd9307..47e2e529874eecec048cfb0b3693c759ec43b5ef 100644
--- a/src/observer/net/server.cpp
+++ b/src/observer/net/server.cpp
@@ -30,6 +30,7 @@ See the Mulan PSL v2 for more details. */
 
 #include "common/lang/mutex.h"
 #include "common/log/log.h"
+#include "common/io/io.h"
 #include "common/seda/seda_config.h"
 #include "event/session_event.h"
 #include "session/session.h"
@@ -189,19 +190,13 @@ int Server::send(ConnectionContext *client, const char *buf, int data_len)
 
   MUTEX_LOCK(&client->mutex);
   int wlen = 0;
-  for (int i = 0; i < 3 && wlen < data_len; i++) {
-    int len = write(client->fd, buf + wlen, data_len - wlen);
-    if (len < 0) {
-      LOG_ERROR("Failed to send data back to client\n");
-      MUTEX_UNLOCK(&client->mutex);
-
-      close_connection(client);
-      return -STATUS_FAILED_NETWORK;
-    }
-    wlen += len;
-  }
-  if (wlen < data_len) {
-    LOG_WARN("Not all data has been send back to client");
+  int ret = common::writen(client->fd, buf, data_len);
+  if (ret < 0) {
+    LOG_ERROR("Failed to send data back to client. ret=%d, error=%s", ret, strerror(errno));
+    MUTEX_UNLOCK(&client->mutex);
+
+    close_connection(client);
+    return -STATUS_FAILED_NETWORK;
   }
 
   MUTEX_UNLOCK(&client->mutex);
diff --git a/src/observer/sql/executor/execute_stage.cpp b/src/observer/sql/executor/execute_stage.cpp
index 5cb4588931cf7989694c3b7bd9d56916b740d964..25a265f01217c7032cda30129c36a56379ecb486 100644
--- a/src/observer/sql/executor/execute_stage.cpp
+++ b/src/observer/sql/executor/execute_stage.cpp
@@ -147,6 +147,9 @@ void ExecuteStage::handle_request(common::StageEvent *event)
     case StmtType::DELETE: {
       do_delete(sql_event);
     } break;
+    default: {
+      LOG_WARN("should not happen. please implenment");
+    } break;
     }
   } else {
     switch (sql->flag) {
diff --git a/src/observer/sql/expr/tuple.h b/src/observer/sql/expr/tuple.h
index 7efed51faf976edab5419b983b6107732d5c6708..dff7751e22f0571306b66cdaafdf3e14610d1e1d 100644
--- a/src/observer/sql/expr/tuple.h
+++ b/src/observer/sql/expr/tuple.h
@@ -21,7 +21,7 @@ See the Mulan PSL v2 for more details. */
 #include "sql/parser/parse.h"
 #include "sql/expr/tuple_cell.h"
 #include "sql/expr/expression.h"
-#include "storage/common/record.h"
+#include "storage/record/record.h"
 
 class Table;
 
@@ -105,7 +105,7 @@ public:
 
   RC cell_at(int index, TupleCell &cell) const override
   {
-    if (index < 0 || index >= speces_.size()) {
+    if (index < 0 || index >= static_cast<int>(speces_.size())) {
       LOG_WARN("invalid argument. index=%d", index);
       return RC::INVALID_ARGUMENT;
     }
@@ -127,7 +127,7 @@ public:
     }
 
     const char *field_name = field.field_name();
-    for (int i = 0; i < speces_.size(); ++i) {
+    for (size_t i = 0; i < speces_.size(); ++i) {
       const FieldExpr * field_expr = (const FieldExpr *)speces_[i]->expression();
       const Field &field = field_expr->field();
       if (0 == strcmp(field_name, field.field_name())) {
@@ -139,7 +139,7 @@ public:
 
   RC cell_spec_at(int index, const TupleCellSpec *&spec) const override
   {
-    if (index < 0 || index >= speces_.size()) {
+    if (index < 0 || index >= static_cast<int>(speces_.size())) {
       LOG_WARN("invalid argument. index=%d", index);
       return RC::INVALID_ARGUMENT;
     }
@@ -202,7 +202,7 @@ public:
 
   RC cell_at(int index, TupleCell &cell) const override
   {
-    if (index < 0 || index >= speces_.size()) {
+    if (index < 0 || index >= static_cast<int>(speces_.size())) {
       return RC::GENERIC_ERROR;
     }
     if (tuple_ == nullptr) {
@@ -219,7 +219,7 @@ public:
   }
   RC cell_spec_at(int index, const TupleCellSpec *&spec) const override
   {
-    if (index < 0 || index >= speces_.size()) {
+    if (index < 0 || index >= static_cast<int>(speces_.size())) {
       return RC::NOTFOUND;
     }
     spec = speces_[index];
diff --git a/src/observer/sql/operator/delete_operator.cpp b/src/observer/sql/operator/delete_operator.cpp
index 88f88c703395fafae246ba0ee382b31a066ef1d1..45e6363bd641aadb400af32c39d4181d9f026826 100644
--- a/src/observer/sql/operator/delete_operator.cpp
+++ b/src/observer/sql/operator/delete_operator.cpp
@@ -14,7 +14,7 @@ See the Mulan PSL v2 for more details. */
 
 #include "common/log/log.h"
 #include "sql/operator/delete_operator.h"
-#include "storage/common/record.h"
+#include "storage/record/record.h"
 #include "storage/common/table.h"
 #include "sql/stmt/delete_stmt.h"
 
diff --git a/src/observer/sql/operator/predicate_operator.cpp b/src/observer/sql/operator/predicate_operator.cpp
index 2874263227d6353e52fb2e96d0df20b8706c0bd9..c9befd3d705ae26d21d20d3fb6d9e0d74dee6399 100644
--- a/src/observer/sql/operator/predicate_operator.cpp
+++ b/src/observer/sql/operator/predicate_operator.cpp
@@ -14,7 +14,7 @@ See the Mulan PSL v2 for more details. */
 
 #include "common/log/log.h"
 #include "sql/operator/predicate_operator.h"
-#include "storage/common/record.h"
+#include "storage/record/record.h"
 #include "sql/stmt/filter_stmt.h"
 #include "storage/common/field.h"
 
diff --git a/src/observer/sql/operator/project_operator.cpp b/src/observer/sql/operator/project_operator.cpp
index 3d7878264ae05099aa03a73fb73972d7355f924f..ed9ea62b89815dd665288ec3eae1bd440abd4946 100644
--- a/src/observer/sql/operator/project_operator.cpp
+++ b/src/observer/sql/operator/project_operator.cpp
@@ -14,7 +14,7 @@ See the Mulan PSL v2 for more details. */
 
 #include "common/log/log.h"
 #include "sql/operator/project_operator.h"
-#include "storage/common/record.h"
+#include "storage/record/record.h"
 #include "storage/common/table.h"
 
 RC ProjectOperator::open()
diff --git a/src/observer/sql/operator/table_scan_operator.h b/src/observer/sql/operator/table_scan_operator.h
index 5d7ef81e42a07b60210d348de4a597bf7a0d4862..a240149df961a99d930f303d412b091d0a214584 100644
--- a/src/observer/sql/operator/table_scan_operator.h
+++ b/src/observer/sql/operator/table_scan_operator.h
@@ -15,7 +15,7 @@ See the Mulan PSL v2 for more details. */
 #pragma once
 
 #include "sql/operator/operator.h"
-#include "storage/common/record_manager.h"
+#include "storage/record/record_manager.h"
 #include "rc.h"
 
 class Table;
diff --git a/src/observer/storage/common/condition_filter.cpp b/src/observer/storage/common/condition_filter.cpp
index 87412360a36fe0d28822e59984c753a1f727d738..749cfa6f3084d7c272cb80aa2707cb7d9aa53fc2 100644
--- a/src/observer/storage/common/condition_filter.cpp
+++ b/src/observer/storage/common/condition_filter.cpp
@@ -14,7 +14,7 @@ See the Mulan PSL v2 for more details. */
 
 #include <stddef.h>
 #include "condition_filter.h"
-#include "record_manager.h"
+#include "storage/record/record_manager.h"
 #include "common/log/log.h"
 #include "storage/common/table.h"
 
diff --git a/src/observer/storage/common/table.cpp b/src/observer/storage/common/table.cpp
index f53be21701492c6603fef6ffc600187e02481364..57e600fb3571a877e0deb5fb3a36a54aa81e4532 100644
--- a/src/observer/storage/common/table.cpp
+++ b/src/observer/storage/common/table.cpp
@@ -22,7 +22,7 @@ See the Mulan PSL v2 for more details. */
 #include "common/log/log.h"
 #include "common/lang/string.h"
 #include "storage/default/disk_buffer_pool.h"
-#include "storage/common/record_manager.h"
+#include "storage/record/record_manager.h"
 #include "storage/common/condition_filter.h"
 #include "storage/common/meta_util.h"
 #include "storage/index/index.h"
diff --git a/src/observer/storage/default/default_handler.cpp b/src/observer/storage/default/default_handler.cpp
index f06e07608ec6f0b6d32ab860889d734d18ec9e88..9e59ff510a38abeaf055dac830ea6e5df195c5d9 100644
--- a/src/observer/storage/default/default_handler.cpp
+++ b/src/observer/storage/default/default_handler.cpp
@@ -19,7 +19,7 @@ See the Mulan PSL v2 for more details. */
 #include "common/os/path.h"
 #include "common/log/log.h"
 #include "common/lang/string.h"
-#include "storage/common/record_manager.h"
+#include "storage/record/record_manager.h"
 #include "storage/index/bplus_tree.h"
 #include "storage/common/table.h"
 #include "storage/common/condition_filter.h"
diff --git a/src/observer/storage/default/disk_buffer_pool.cpp b/src/observer/storage/default/disk_buffer_pool.cpp
index 18a24508acb486e2aa9b9d2c5d0dea977bb63a20..7b265d6a3f5c32dae33e595b03dcd4e1c56a65d9 100644
--- a/src/observer/storage/default/disk_buffer_pool.cpp
+++ b/src/observer/storage/default/disk_buffer_pool.cpp
@@ -18,11 +18,12 @@ See the Mulan PSL v2 for more details. */
 #include "common/lang/mutex.h"
 #include "common/log/log.h"
 #include "common/os/os.h"
+#include "common/io/io.h"
 
 using namespace common;
 
 static const PageNum BP_HEADER_PAGE = 0;
-static const int MEM_POOL_ITEM_NUM = 50;
+static const int MEM_POOL_ITEM_NUM = 128;
 
 unsigned long current_time()
 {
@@ -31,61 +32,102 @@ unsigned long current_time()
   return tp.tv_sec * 1000 * 1000 * 1000UL + tp.tv_nsec;
 }
 
-BPFrameManager::BPFrameManager(const char *name) : MemPoolSimple<Frame>(name)
+BPFrameManager::BPFrameManager(const char *name) : allocator_(name)
 {}
 
-static bool match_purge(void *item, void *arg)
+RC BPFrameManager::init(int pool_num)
 {
-  Frame *frame = (Frame *)item;
-  return frame->can_purge();
+  int ret =  allocator_.init(false, pool_num);
+  if (ret == 0) {
+    return RC::SUCCESS;
+  }
+  return RC::GENERIC_ERROR;
 }
 
-Frame *BPFrameManager::begin_purge()
+RC BPFrameManager::cleanup()
 {
-  return MemPoolSimple<Frame>::find(match_purge, nullptr);
-}
-
-struct MatchFilePage {
-  MatchFilePage(int file_desc, PageNum page_num)
-  {
-    this->file_desc = file_desc;
-    this->page_num = page_num;
+  if (frames_.count() > 0) {
+    return RC::GENERIC_ERROR;
   }
-  int file_desc;
-  PageNum page_num;
-};
 
-static bool match_file_page(void *item, void *arg)
+  frames_.destroy();
+  return RC::SUCCESS;
+}
+
+Frame *BPFrameManager::begin_purge()
 {
-  Frame *frame = (Frame *)item;
-  MatchFilePage *file_page = (MatchFilePage *)arg;
+  Frame *frame_can_purge = nullptr;
+  auto purge_finder = [&frame_can_purge](const BPFrameId &frame_id, Frame * const frame) {
+    if (frame->can_purge()) {
+      frame_can_purge = frame;
+      return false; // false to break the progress
+    }
+    return true; // true continue to look up
+  };
+  frames_.foreach_reverse(purge_finder);
+  return frame_can_purge;
+}
 
-  if (frame->file_desc() == file_page->file_desc && frame->page_num() == file_page->page_num)
-    return true;
+Frame *BPFrameManager::get(int file_desc, PageNum page_num)
+{
+  BPFrameId frame_id(file_desc, page_num);
 
-  return false;
+  std::lock_guard<std::mutex> lock_guard(lock_);
+  Frame *frame = nullptr;
+  (void)frames_.get(frame_id, frame);
+  return frame;
 }
 
-Frame *BPFrameManager::get(int file_desc, PageNum page_num)
+Frame *BPFrameManager::alloc(int file_desc, PageNum page_num)
 {
-  MatchFilePage file_page(file_desc, page_num);
-  return MemPoolSimple<Frame>::find(match_file_page, &file_page);
+  BPFrameId frame_id(file_desc, page_num);
+
+  std::lock_guard<std::mutex> lock_guard(lock_);
+  Frame *frame = nullptr;
+  bool found = frames_.get(frame_id, frame);
+  if (found) {
+    // assert (frame != nullptr);
+    return nullptr; // should use get
+  }
+
+  frame = allocator_.alloc();
+  if (frame != nullptr) {
+    frames_.put(frame_id, frame);
+  }
+  return frame;
 }
 
-static bool match_file(void *item, void *arg)
+RC BPFrameManager::free(int file_desc, PageNum page_num, Frame *frame)
 {
-  Frame *frame = (Frame *)item;
-  int *file_desc = (int *)arg;
+  BPFrameId frame_id(file_desc, page_num);
 
-  if (frame->file_desc() == *file_desc)
-    return true;
+  std::lock_guard<std::mutex> lock_guard(lock_);
+  Frame *frame_source = nullptr;
+  bool found = frames_.get(frame_id, frame_source);
+  if (!found || frame != frame_source) {
+    LOG_WARN("failed to find frame or got frame not match. file_desc=%d, PageNum=%d, frame_source=%p, frame=%p",
+             file_desc, page_num, frame_source, frame);
+    return RC::GENERIC_ERROR;
+  }
 
-  return false;
+  frames_.remove(frame_id);
+  allocator_.free(frame);
+  return RC::SUCCESS;
 }
 
 std::list<Frame *> BPFrameManager::find_list(int file_desc)
 {
-  return find_all(match_file, &file_desc);
+  std::lock_guard<std::mutex> lock_guard(lock_);
+
+  std::list<Frame *> frames;
+  auto fetcher = [&frames, file_desc](const BPFrameId &frame_id, Frame * const frame) -> bool {
+    if (file_desc == frame_id.file_desc()) {
+      frames.push_back(frame);
+    }
+    return true;
+  };
+  frames_.foreach(fetcher);
+  return frames;
 }
 
 ////////////////////////////////////////////////////////////////////////////////
@@ -149,7 +191,7 @@ RC DiskBufferPool::open_file(const char *file_name)
   file_desc_ = fd;
 
   RC rc = RC::SUCCESS;
-  rc = allocate_frame(&hdr_frame_);
+  rc = allocate_frame(BP_HEADER_PAGE, &hdr_frame_);
   if (rc != RC::SUCCESS) {
     LOG_ERROR("failed to allocate frame for header. file name %s", file_name_.c_str());
     close(fd);
@@ -164,7 +206,7 @@ RC DiskBufferPool::open_file(const char *file_name)
   if ((rc = load_page(BP_HEADER_PAGE, hdr_frame_)) != RC::SUCCESS) {
     LOG_ERROR("Failed to load first page of %s, due to %s.", file_name, strerror(errno));
     hdr_frame_->pin_count_ = 0;
-    purge_frame(hdr_frame_);
+    purge_frame(BP_HEADER_PAGE, hdr_frame_);
     close(fd);
     file_desc_ = -1;
     return rc;
@@ -212,15 +254,13 @@ RC DiskBufferPool::get_this_page(PageNum page_num, Frame **frame)
     used_match_frame->pin_count_++;
     used_match_frame->acc_time_ = current_time();
 
-    frame_manager_.mark_modified(used_match_frame);
-
     *frame = used_match_frame;
     return RC::SUCCESS;
   }
 
   // Allocate one page and load the data into this page
   Frame *allocated_frame = nullptr;
-  if ((rc = allocate_frame(&allocated_frame)) != RC::SUCCESS) {
+  if ((rc = allocate_frame(page_num, &allocated_frame)) != RC::SUCCESS) {
     LOG_ERROR("Failed to alloc frame %s:%d, due to failed to alloc page.", file_name_.c_str(), page_num);
     return rc;
   }
@@ -232,7 +272,7 @@ RC DiskBufferPool::get_this_page(PageNum page_num, Frame **frame)
   if ((rc = load_page(page_num, allocated_frame)) != RC::SUCCESS) {
     LOG_ERROR("Failed to load page %s:%d", file_name_.c_str(), page_num);
     allocated_frame->pin_count_ = 0;
-    purge_frame(allocated_frame);
+    purge_frame(page_num, allocated_frame);
     return rc;
   }
 
@@ -260,13 +300,19 @@ RC DiskBufferPool::allocate_page(Frame **frame)
     }
   }
 
+  if (file_header_->page_count >= BPFileHeader::MAX_PAGE_NUM) {
+    LOG_WARN("file buffer pool is full. page count %d, max page count %d",
+	     file_header_->page_count, BPFileHeader::MAX_PAGE_NUM);
+    return BUFFERPOOL_NOBUF;
+  }
+
+  PageNum page_num = file_header_->page_count;
   Frame *allocated_frame = nullptr;
-  if ((rc = allocate_frame(&allocated_frame)) != RC::SUCCESS) {
+  if ((rc = allocate_frame(page_num, &allocated_frame)) != RC::SUCCESS) {
     LOG_ERROR("Failed to allocate frame %s, due to no free page.", file_name_.c_str());
     return rc;
   }
 
-  PageNum page_num = file_header_->page_count;
   file_header_->allocated_pages++;
   file_header_->page_count++;
 
@@ -332,7 +378,7 @@ RC DiskBufferPool::dispose_page(PageNum page_num)
   return RC::SUCCESS;
 }
 
-RC DiskBufferPool::purge_frame(Frame *buf)
+RC DiskBufferPool::purge_frame(PageNum page_num, Frame *buf)
 {
   if (buf->pin_count_ > 0) {
     LOG_INFO("Begin to free page %d of %d(file id), but it's pinned, pin_count:%d.",
@@ -349,7 +395,7 @@ RC DiskBufferPool::purge_frame(Frame *buf)
   }
 
   LOG_DEBUG("Successfully purge frame =%p, page %d of %d(file desc)", buf, buf->page_num(), buf->file_desc_);
-  frame_manager_.free(buf);
+  frame_manager_.free(file_desc_, page_num, buf);
   return RC::SUCCESS;
 }
 
@@ -363,7 +409,7 @@ RC DiskBufferPool::purge_page(PageNum page_num)
 {
   Frame *used_frame = frame_manager_.get(file_desc_, page_num);
   if (used_frame != nullptr) {
-    return purge_frame(used_frame);
+    return purge_frame(page_num, used_frame);
   }
 
   return RC::SUCCESS;
@@ -386,7 +432,7 @@ RC DiskBufferPool::purge_all_pages()
         return rc;
       }
     }
-    frame_manager_.free(frame);
+    frame_manager_.free(file_desc_, frame->page_.page_num, frame);
   }
   return RC::SUCCESS;
 }
@@ -419,7 +465,7 @@ RC DiskBufferPool::flush_page(Frame &frame)
     return RC::IOERR_SEEK;
   }
 
-  if (write(file_desc_, &page, sizeof(Page)) != sizeof(Page)) {
+  if (writen(file_desc_, &page, sizeof(Page)) != 0) {
     LOG_ERROR("Failed to flush page %lld of %d due to %s.", offset, file_desc_, strerror(errno));
     return RC::IOERR_WRITE;
   }
@@ -442,32 +488,32 @@ RC DiskBufferPool::flush_all_pages()
   return RC::SUCCESS;
 }
 
-RC DiskBufferPool::allocate_frame(Frame **buffer)
+RC DiskBufferPool::allocate_frame(PageNum page_num, Frame **buffer)
 {
-  Frame *frame = frame_manager_.alloc();
-  if (frame != nullptr) {
-    *buffer = frame;
-    return RC::SUCCESS;
-  }
-
-  frame = frame_manager_.begin_purge();
-  if (frame == nullptr) {
-    LOG_ERROR("All pages have been used and pinned.");
-    return RC::NOMEM;
-  }
+  while (true) {
+    Frame *frame = frame_manager_.alloc(file_desc_, page_num);
+    if (frame != nullptr) {
+      *buffer = frame;
+      return RC::SUCCESS;
+    }
 
-  if (frame->dirty_) {
-    RC rc = bp_manager_.flush_page(*frame);
-    if (rc != RC::SUCCESS) {
-      LOG_ERROR("Failed to aclloc block due to failed to flush old block.");
-      return rc;
+    frame = frame_manager_.begin_purge();
+    if (frame == nullptr) {
+      LOG_ERROR("All pages have been used and pinned.");
+      return RC::NOMEM;
     }
-  }
 
-  frame_manager_.mark_modified(frame);
+    if (frame->dirty_) {
+      RC rc = bp_manager_.flush_page(*frame);
+      if (rc != RC::SUCCESS) {
+        LOG_ERROR("Failed to aclloc block due to failed to flush old block.");
+        return rc;
+      }
+    }
 
-  *buffer = frame;
-  return RC::SUCCESS;
+    frame_manager_.free(frame->file_desc(), frame->page_num(), frame);
+  }
+  return RC::INTERNAL;
 }
 
 RC DiskBufferPool::check_page_num(PageNum page_num)
@@ -492,9 +538,11 @@ RC DiskBufferPool::load_page(PageNum page_num, Frame *frame)
 
     return RC::IOERR_SEEK;
   }
-  if (read(file_desc_, &(frame->page_), sizeof(Page)) != sizeof(Page)) {
-    LOG_ERROR("Failed to load page %s:%d, due to failed to read data:%s.",
-	      file_name_.c_str(), page_num, strerror(errno));
+
+  int ret = readn(file_desc_, &(frame->page_), sizeof(Page));
+  if (ret != 0) {
+    LOG_ERROR("Failed to load page %s:%d, due to failed to read data:%s, ret=%d, page count=%d",
+	      file_name_.c_str(), page_num, strerror(errno), ret, file_header_->allocated_pages);
     return RC::IOERR_READ;
   }
   return RC::SUCCESS;
@@ -512,7 +560,7 @@ int DiskBufferPool::file_desc() const
 ////////////////////////////////////////////////////////////////////////////////
 BufferPoolManager::BufferPoolManager()
 {
-  frame_manager_.init(true/*dynamic*/, MEM_POOL_ITEM_NUM);
+  frame_manager_.init(MEM_POOL_ITEM_NUM);
 }
 
 BufferPoolManager::~BufferPoolManager()
@@ -559,7 +607,7 @@ RC BufferPoolManager::create_file(const char *file_name)
     return RC::IOERR_SEEK;
   }
 
-  if (write(fd, (char *)&page, sizeof(Page)) != sizeof(Page)) {
+  if (writen(fd, (char *)&page, sizeof(Page)) != 0) {
     LOG_ERROR("Failed to write header to file %s, due to %s.", file_name, strerror(errno));
     close(fd);
     return RC::IOERR_WRITE;
diff --git a/src/observer/storage/default/disk_buffer_pool.h b/src/observer/storage/default/disk_buffer_pool.h
index 06c89b399372416f60d0300de32b439cc5fde0c2..5e7245327313109b771a8c1610d6ba104b16f751 100644
--- a/src/observer/storage/default/disk_buffer_pool.h
+++ b/src/observer/storage/default/disk_buffer_pool.h
@@ -11,8 +11,7 @@ See the Mulan PSL v2 for more details. */
 //
 // Created by Meiyi & Longda on 2021/4/13.
 //
-#ifndef __OBSERVER_STORAGE_COMMON_PAGE_MANAGER_H_
-#define __OBSERVER_STORAGE_COMMON_PAGE_MANAGER_H_
+#pragma once
 
 #include <sys/types.h>
 #include <sys/stat.h>
@@ -22,11 +21,13 @@ See the Mulan PSL v2 for more details. */
 #include <string.h>
 #include <time.h>
 #include <string>
+#include <mutex>
 #include <unordered_map>
 
 #include "rc.h"
 #include "defs.h"
 #include "common/mm/mem_pool.h"
+#include "common/lang/lru_cache.h"
 #include "common/lang/bitmap.h"
 
 class BufferPoolManager;
@@ -34,10 +35,9 @@ class DiskBufferPool;
 
 //
 #define BP_INVALID_PAGE_NUM (-1)
-#define BP_PAGE_SIZE (1 << 13)
+#define BP_PAGE_SIZE (1 << 14)
 #define BP_PAGE_DATA_SIZE (BP_PAGE_SIZE - sizeof(PageNum))
 #define BP_FILE_SUB_HDR_SIZE (sizeof(BPFileSubHeader))
-#define BP_BUFFER_SIZE 256
 
 struct Page {
   PageNum page_num;
@@ -116,20 +116,82 @@ private:
   Page          page_;
 };
 
-class BPFrameManager : public common::MemPoolSimple<Frame>
+class BPFrameId
+{
+public: 
+  BPFrameId(int file_desc, PageNum page_num) :
+    file_desc_(file_desc), page_num_(page_num)
+  {}
+
+  bool equal_to(const BPFrameId &other) const
+  {
+    return file_desc_ == other.file_desc_ && page_num_ == other.page_num_;
+  }
+
+  bool operator== (const BPFrameId &other) const
+  {
+    return this->equal_to(other);
+  }
+
+  size_t hash() const
+  {
+    return static_cast<size_t>(file_desc_) << 32L | page_num_;
+  }
+
+  int file_desc() const { return file_desc_; }
+  PageNum page_num() const { return page_num_; }
+
+private:
+  int file_desc_;
+  PageNum page_num_;
+};
+
+class BPFrameManager
 {
 public:
   BPFrameManager(const char *tag);
 
+  RC init(int pool_num);
+  RC cleanup();
+
   Frame *get(int file_desc, PageNum page_num);
 
   std::list<Frame *> find_list(int file_desc);
 
+  Frame *alloc(int file_desc, PageNum page_num);
+
+  /**
+   * 尽管frame中已经包含了file_desc和page_num，但是依然要求
+   * 传入，因为frame可能忘记初始化或者没有初始化
+   */
+  RC free(int file_desc, PageNum page_num, Frame *frame);
+
   /**
    * 如果不能从空闲链表中分配新的页面，就使用这个接口，
    * 尝试从pin count=0的页面中淘汰一个
    */
   Frame *begin_purge();
+
+  size_t frame_num() const { return frames_.count(); }
+
+  /**
+   * 测试使用。返回已经从内存申请的个数
+   */
+  size_t total_frame_num() const { return allocator_.get_size(); }
+
+private:
+  class BPFrameIdHasher {
+  public:
+    size_t operator() (const BPFrameId &frame_id) const {
+      return frame_id.hash();
+    }
+  };
+  using FrameLruCache = common::LruCache<BPFrameId, Frame *, BPFrameIdHasher>;
+  using FrameAllocator = common::MemPoolSimple<Frame>;
+
+  std::mutex lock_;
+  FrameLruCache frames_;
+  FrameAllocator allocator_;
 };
 
 class BufferPoolIterator
@@ -223,12 +285,12 @@ public:
   RC flush_all_pages();
 
 protected:
-  RC allocate_frame(Frame **buf);
+  RC allocate_frame(PageNum page_num, Frame **buf);
 
   /**
    * 刷新指定页面到磁盘(flush)，并且释放关联的Frame
    */
-  RC purge_frame(Frame *used_frame);
+  RC purge_frame(PageNum page_num, Frame *used_frame);
   RC check_page_num(PageNum page_num);
 
   /**
@@ -270,5 +332,3 @@ private:
   std::unordered_map<std::string, DiskBufferPool *> buffer_pools_;
   std::unordered_map<int, DiskBufferPool *> fd_buffer_pools_;
 };
-
-#endif  //__OBSERVER_STORAGE_COMMON_PAGE_MANAGER_H_
diff --git a/src/observer/storage/index/bplus_tree.cpp b/src/observer/storage/index/bplus_tree.cpp
index 0e7a62d68cf3eb69273195fdee001b972ea30935..41a255ca8cb3042d5f3d4e6d36d3748fc84ed948 100644
--- a/src/observer/storage/index/bplus_tree.cpp
+++ b/src/observer/storage/index/bplus_tree.cpp
@@ -17,6 +17,7 @@ See the Mulan PSL v2 for more details. */
 #include "rc.h"
 #include "common/log/log.h"
 #include "sql/parser/parse_defs.h"
+#include "common/lang/lower_bound.h"
 
 #define FIRST_INDEX_PAGE 1
 
@@ -176,23 +177,10 @@ int LeafIndexNodeHandler::min_size() const
 int LeafIndexNodeHandler::lookup(const KeyComparator &comparator, const char *key, bool *found /* = nullptr */) const
 {
   const int size = this->size();
-  int i = 0;
-  for ( ; i < size; i++) {
-    int result = comparator(key, __key_at(i));
-    if (0 == result) {
-      if (found) {
-	*found = true;
-      }
-      return i;
-    }
-    if (result < 0) {
-      break;
-    }
-  }
-  if (found) {
-    *found = false;
-  }
-  return i;
+  common::BinaryIterator<char> iter_begin(item_size(), __key_at(0));
+  common::BinaryIterator<char> iter_end(item_size(), __key_at(size));
+  common::BinaryIterator<char> iter = lower_bound(iter_begin, iter_end, key, comparator, found);
+  return iter - iter_begin;
 }
 
 void LeafIndexNodeHandler::insert(int index, const char *key, const char *value)
@@ -468,37 +456,29 @@ int InternalIndexNodeHandler::min_size() const
 int InternalIndexNodeHandler::lookup(const KeyComparator &comparator, const char *key,
 				     bool *found /* = nullptr */, int *insert_position /*= nullptr */) const
 {
-  int i = 1;
   const int size = this->size();
-  for ( ; i < size; i++) {
-    int result = comparator(key, __key_at(i));
-    if (result == 0) {
-      if (found) {
-	*found = true;
-      }
-      if (insert_position) {
-	*insert_position = i;
-      }
-      return i;
+  if (size == 0) {
+    if (insert_position) {
+      *insert_position = 1;
     }
-    if (result < 0) {
-      if (found) {
-	*found = false;
-      }
-      if (insert_position) {
-	*insert_position = i;
-      }
-
-      return i - 1;
+    if (found) {
+      *found = false;
     }
+    return 0;
   }
-  if (found) {
-    *found = false;
-  }
+
+  common::BinaryIterator<char> iter_begin(item_size(), __key_at(1));
+  common::BinaryIterator<char> iter_end(item_size(), __key_at(size));
+  common::BinaryIterator<char> iter = lower_bound(iter_begin, iter_end, key, comparator, found);
+  int ret = static_cast<int>(iter - iter_begin) + 1;
   if (insert_position) {
-    *insert_position = size;
+    *insert_position = ret;
+  }
+
+  if (ret >= size || comparator(key, __key_at(ret)) < 0) {
+    return ret - 1;
   }
-  return size - 1;
+  return ret;
 }
 
 char *InternalIndexNodeHandler::key_at(int index)
@@ -1049,6 +1029,7 @@ bool BplusTreeHandler::validate_leaf_link()
   while (result && next_page_num != BP_INVALID_PAGE_NUM) {
     rc = disk_buffer_pool_->get_this_page(next_page_num, &frame);
     if (rc != RC::SUCCESS) {
+      free_key(prev_key);
       LOG_WARN("failed to fetch next page. page num=%d, rc=%d:%s", next_page_num, rc, strrc(rc));
       return false;
     }
@@ -1317,12 +1298,6 @@ RC BplusTreeHandler::split(Frame *frame, Frame *&new_frame)
 {
   IndexNodeHandlerType old_node(file_header_, frame);
 
-  char *new_parent_key = (char *)mem_pool_item_->alloc();
-  if (new_parent_key == nullptr) {
-    LOG_WARN("Failed to alloc memory for new key. size=%d", file_header_.key_length);
-    return RC::NOMEM;
-  }
-
   // add a new node
   RC rc = disk_buffer_pool_->allocate_page(&new_frame);
   if (rc != RC::SUCCESS) {
@@ -1417,7 +1392,9 @@ RC BplusTreeHandler::insert_entry(const char *user_key, const RID *rid)
   }
 
   if (is_empty()) {
-    return create_new_tree(key, rid);
+    RC rc = create_new_tree(key, rid);
+    mem_pool_item_->free(key);
+    return rc;
   }
 
   Frame *frame;
diff --git a/src/observer/storage/index/bplus_tree.h b/src/observer/storage/index/bplus_tree.h
index e8a2d01a73427b5902a5a1b0496a43e27645fcfd..cd946462889e330cb032a60b5f806438b0ea2947 100644
--- a/src/observer/storage/index/bplus_tree.h
+++ b/src/observer/storage/index/bplus_tree.h
@@ -21,7 +21,7 @@ See the Mulan PSL v2 for more details. */
 #include <sstream>
 #include <functional>
 
-#include "storage/common/record_manager.h"
+#include "storage/record/record_manager.h"
 #include "storage/default/disk_buffer_pool.h"
 #include "sql/parser/parse_defs.h"
 #include "util/comparator.h"
diff --git a/src/observer/storage/index/index.h b/src/observer/storage/index/index.h
index 4f04fb4d3c2bf5d96409c670f11ff48ee98c66d8..165b8414887a199476863755a737f550e1a99583 100644
--- a/src/observer/storage/index/index.h
+++ b/src/observer/storage/index/index.h
@@ -21,7 +21,7 @@ See the Mulan PSL v2 for more details. */
 #include "rc.h"
 #include "storage/common/index_meta.h"
 #include "storage/common/field_meta.h"
-#include "storage/common/record_manager.h"
+#include "storage/record/record_manager.h"
 
 class IndexDataOperator {
 public:
diff --git a/src/observer/storage/common/record.h b/src/observer/storage/record/record.h
similarity index 100%
rename from src/observer/storage/common/record.h
rename to src/observer/storage/record/record.h
diff --git a/src/observer/storage/common/record_manager.cpp b/src/observer/storage/record/record_manager.cpp
similarity index 91%
rename from src/observer/storage/common/record_manager.cpp
rename to src/observer/storage/record/record_manager.cpp
index 4a4b4b998a3aa79b882fc77a46c1315189f35236..8a0506ad9f3fd290b515a689b3f930b335cc1aba 100644
--- a/src/observer/storage/common/record_manager.cpp
+++ b/src/observer/storage/record/record_manager.cpp
@@ -11,11 +11,11 @@ See the Mulan PSL v2 for more details. */
 //
 // Created by Meiyi & Longda on 2021/4/13.
 //
-#include "storage/common/record_manager.h"
+#include "storage/record/record_manager.h"
 #include "rc.h"
 #include "common/log/log.h"
 #include "common/lang/bitmap.h"
-#include "condition_filter.h"
+#include "storage/common/condition_filter.h"
 
 using namespace common;
 
@@ -262,7 +262,9 @@ RC RecordFileHandler::init(DiskBufferPool *buffer_pool)
 
   disk_buffer_pool_ = buffer_pool;
 
-  LOG_INFO("Successfully open record file handle");
+  RC rc = init_free_pages();
+
+  LOG_INFO("open record file handle done. rc=%s", strrc(rc));
   return RC::SUCCESS;
 }
 
@@ -273,18 +275,42 @@ void RecordFileHandler::close()
   }
 }
 
-RC RecordFileHandler::insert_record(const char *data, int record_size, RID *rid)
+RC RecordFileHandler::init_free_pages()
 {
-  RC ret = RC::SUCCESS;
-  // 找到没有填满的页面
+  // 遍历当前文件上所有页面，找到没有满的页面
+  // 这个效率很低，会降低启动速度
 
+  RC rc = RC::SUCCESS;
   BufferPoolIterator bp_iterator;
   bp_iterator.init(*disk_buffer_pool_);
   RecordPageHandler record_page_handler;
-  bool page_found = false;
   PageNum current_page_num = 0;
   while (bp_iterator.has_next()) {
     current_page_num = bp_iterator.next();
+    rc = record_page_handler.init(*disk_buffer_pool_, current_page_num);
+    if (rc != RC::SUCCESS) {
+      LOG_WARN("failed to init record page handler. page num=%d, rc=%d:%s", current_page_num, rc, strrc(rc));
+      return rc;
+    }
+
+    if (!record_page_handler.is_full()) {
+      free_pages_.insert(current_page_num);
+    }
+    record_page_handler.cleanup();
+  }
+  return rc;
+}
+
+RC RecordFileHandler::insert_record(const char *data, int record_size, RID *rid)
+{
+  RC ret = RC::SUCCESS;
+  // 找到没有填满的页面
+
+  RecordPageHandler record_page_handler;
+  bool page_found = false;
+  PageNum current_page_num = 0;
+  while (!free_pages_.empty()) {
+    current_page_num = *free_pages_.begin();
     ret = record_page_handler.init(*disk_buffer_pool_, current_page_num);
     if (ret != RC::SUCCESS) {
       LOG_WARN("failed to init record page handler. page num=%d, rc=%d:%s", current_page_num, ret, strrc(ret));
@@ -296,6 +322,7 @@ RC RecordFileHandler::insert_record(const char *data, int record_size, RID *rid)
       break;
     }
     record_page_handler.cleanup();
+    free_pages_.erase(free_pages_.begin());
   }
 
   // 找不到就分配一个新的页面
@@ -317,6 +344,7 @@ RC RecordFileHandler::insert_record(const char *data, int record_size, RID *rid)
     }
 
     disk_buffer_pool_->unpin_page(frame);
+    free_pages_.insert(current_page_num);
   }
 
   // 找到空闲位置
@@ -337,13 +365,17 @@ RC RecordFileHandler::update_record(const Record *rec)
 
 RC RecordFileHandler::delete_record(const RID *rid)
 {
-  RC ret = RC::SUCCESS;
+  RC rc = RC::SUCCESS;
   RecordPageHandler page_handler;
-  if ((ret != page_handler.init(*disk_buffer_pool_, rid->page_num)) != RC::SUCCESS) {
-    LOG_ERROR("Failed to init record page handler.page number=%d", rid->page_num);
-    return ret;
+  if ((rc != page_handler.init(*disk_buffer_pool_, rid->page_num)) != RC::SUCCESS) {
+    LOG_ERROR("Failed to init record page handler.page number=%d. rc=%s", rid->page_num, strrc(rc));
+    return rc;
   }
-  return page_handler.delete_record(rid);
+  rc = page_handler.delete_record(rid);
+  if (rc == RC::SUCCESS) {
+    free_pages_.insert(rid->page_num);
+  }
+  return rc;
 }
 
 RC RecordFileHandler::get_record(const RID *rid, Record *rec)
diff --git a/src/observer/storage/common/record_manager.h b/src/observer/storage/record/record_manager.h
similarity index 93%
rename from src/observer/storage/common/record_manager.h
rename to src/observer/storage/record/record_manager.h
index c9991d12c83a4c47d54b670b4f3ab5deb761bfb7..c858bcfe8bc5b12888218e92765531a6d04fb4bf 100644
--- a/src/observer/storage/common/record_manager.h
+++ b/src/observer/storage/record/record_manager.h
@@ -11,13 +11,12 @@ See the Mulan PSL v2 for more details. */
 //
 // Created by Meiyi & Longda on 2021/4/13.
 //
-#ifndef __OBSERVER_STORAGE_COMMON_RECORD_MANAGER_H_
-#define __OBSERVER_STORAGE_COMMON_RECORD_MANAGER_H_
+#pragma once
 
 #include <sstream>
 #include <limits>
 #include "storage/default/disk_buffer_pool.h"
-#include "storage/common/record.h"
+#include "storage/record/record.h"
 #include "common/lang/bitmap.h"
 
 class ConditionFilter;
@@ -30,15 +29,6 @@ struct PageHeader {
   int32_t first_record_offset;  // 第一条记录的偏移量
 };
 
-
-class RidDigest {
-public:
-  size_t operator()(const RID &rid) const
-  {
-    return ((size_t)(rid.page_num) << 32) | rid.slot_num;
-  }
-};
-
 class RecordPageHandler;
 class RecordPageIterator
 {
@@ -149,8 +139,12 @@ public:
     return page_handler.update_record_in_place(rid, updater);
   }
 
+private:
+  RC init_free_pages();
+  
 private:
   DiskBufferPool *disk_buffer_pool_ = nullptr;
+  std::unordered_set<PageNum>  free_pages_; // 没有填充满的页面集合
 };
 
 class RecordFileScanner {
@@ -183,5 +177,3 @@ private:
   RecordPageIterator record_page_iterator_;
   Record next_record_;
 };
-
-#endif  //__OBSERVER_STORAGE_COMMON_RECORD_MANAGER_H_
diff --git a/src/observer/storage/trx/trx.cpp b/src/observer/storage/trx/trx.cpp
index fada44eb84a164f79df976997d79579b75c6f9aa..304cda8179c454151542ea2b269ccca02836b7de 100644
--- a/src/observer/storage/trx/trx.cpp
+++ b/src/observer/storage/trx/trx.cpp
@@ -16,7 +16,7 @@ See the Mulan PSL v2 for more details. */
 
 #include "storage/trx/trx.h"
 #include "storage/common/table.h"
-#include "storage/common/record_manager.h"
+#include "storage/record/record_manager.h"
 #include "storage/common/field_meta.h"
 #include "common/log/log.h"
 
diff --git a/src/observer/storage/trx/trx.h b/src/observer/storage/trx/trx.h
index be34c0df1d035d4cfaf326a4d7f47f66545a7212..7d7250d59bb9a1c7e3f3159c830a41ba5186b783 100644
--- a/src/observer/storage/trx/trx.h
+++ b/src/observer/storage/trx/trx.h
@@ -21,7 +21,7 @@ See the Mulan PSL v2 for more details. */
 #include <mutex>
 
 #include "sql/parser/parse.h"
-#include "storage/common/record_manager.h"
+#include "storage/record/record_manager.h"
 #include "rc.h"
 
 class Table;
diff --git a/unitest/bp_manager_test.cpp b/unitest/bp_manager_test.cpp
index cef9d4310ebbd66d414ca4cbe48c1549953dec06..f9b87245b6c0284327596f35dcc71c0ba766f793 100644
--- a/unitest/bp_manager_test.cpp
+++ b/unitest/bp_manager_test.cpp
@@ -17,89 +17,101 @@ See the Mulan PSL v2 for more details. */
 
 void test_get(BPFrameManager &frame_manager)
 {
-  Frame *frame1 = frame_manager.alloc();
+  const int file_desc = 0;
+  PageNum page_num = 1;
+  Frame *frame1 = frame_manager.alloc(file_desc, page_num);
   ASSERT_NE(frame1, nullptr);
 
-  frame1->set_file_desc(0);
-  frame1->set_page_num(1);
+  frame1->set_file_desc(file_desc);
+  frame1->set_page_num(page_num);
 
-  ASSERT_EQ(frame1, frame_manager.get(0, 1));
+  ASSERT_EQ(frame1, frame_manager.get(file_desc, 1));
 
-  Frame *frame2 = frame_manager.alloc();
+  Frame *frame2 = frame_manager.alloc(file_desc, 2);
   ASSERT_NE(frame2, nullptr);
   frame2->set_file_desc(0);
   frame2->set_page_num(2);
 
-  ASSERT_EQ(frame1, frame_manager.get(0, 1));
+  ASSERT_EQ(frame1, frame_manager.get(file_desc, 1));
 
-  Frame *frame3 = frame_manager.alloc();
+  Frame *frame3 = frame_manager.alloc(file_desc, 3);
   ASSERT_NE(frame3, nullptr);
-  frame3->set_file_desc(0);
+  frame3->set_file_desc(file_desc);
   frame3->set_page_num(3);
 
-  frame2 = frame_manager.get(0, 2);
+  frame2 = frame_manager.get(file_desc, 2);
   ASSERT_NE(frame2, nullptr);
 
-  Frame *frame4 = frame_manager.alloc();
-  frame4->set_file_desc(0);
+  Frame *frame4 = frame_manager.alloc(file_desc, 4);
+  frame4->set_file_desc(file_desc);
   frame4->set_page_num(4);
 
-  frame_manager.free(frame1);
-  frame1 = frame_manager.get(0, 1);
+  frame_manager.free(file_desc, frame1->page_num(), frame1);
+  frame1 = frame_manager.get(file_desc, 1);
   ASSERT_EQ(frame1, nullptr);
 
-  ASSERT_EQ(frame3, frame_manager.get(0, 3));
+  ASSERT_EQ(frame3, frame_manager.get(file_desc, 3));
 
-  ASSERT_EQ(frame4, frame_manager.get(0, 4));
+  ASSERT_EQ(frame4, frame_manager.get(file_desc, 4));
 
-  frame_manager.free(frame2);
-  frame_manager.free(frame3);
-  frame_manager.free(frame4);
+  frame_manager.free(file_desc, frame2->page_num(), frame2);
+  frame_manager.free(file_desc, frame3->page_num(), frame3);
+  frame_manager.free(file_desc, frame4->page_num(), frame4);
 
-  ASSERT_EQ(nullptr, frame_manager.get(0, 2));
-  ASSERT_EQ(nullptr, frame_manager.get(0, 3));
-  ASSERT_EQ(nullptr, frame_manager.get(0, 4));
+  ASSERT_EQ(nullptr, frame_manager.get(file_desc, 2));
+  ASSERT_EQ(nullptr, frame_manager.get(file_desc, 3));
+  ASSERT_EQ(nullptr, frame_manager.get(file_desc, 4));
 }
 
 void test_alloc(BPFrameManager &frame_manager)
 {
-  int size = frame_manager.get_size();
-
   std::list<Frame *> used_list;
 
-  for (int i = 0; i < size; i++) {
-    Frame *item = frame_manager.alloc();
-    ASSERT_NE(item, nullptr);
-    used_list.push_back(item);
+  const int file_desc = 0;
+  size_t size = 0;
+  for ( ; true; size++) {
+    Frame *item = frame_manager.alloc(file_desc, size);
+    if (item != nullptr) {
+      item->set_file_desc(file_desc);
+      item->set_page_num(size);
+      used_list.push_back(item);
+    } else {
+      break;
+    }
   }
+  size++;
 
-  ASSERT_EQ(used_list.size(), frame_manager.get_used_num());
+  ASSERT_EQ(used_list.size(), frame_manager.frame_num());
 
-  for (int i = 0; i < size; i++) {
-    Frame *item = frame_manager.alloc();
+  for (size_t i = size; i < size * 2; i++) {
+    Frame *item = frame_manager.alloc(file_desc, i);
 
     ASSERT_EQ(item, nullptr);
   }
 
-  for (int i = 0; i < size * 10; i++) {
-    if (i % 2 == 0) {
+  for (size_t i = size * 2; i < size * 10; i++) {
+    if (i % 2 == 0) { // from size * 2, that free one frame first
       Frame *item = used_list.front();
       used_list.pop_front();
 
-      frame_manager.free(item);
+      RC rc = frame_manager.free(file_desc, item->page_num(), item);
+      ASSERT_EQ(rc, RC::SUCCESS);
     } else {
-      Frame *item = frame_manager.alloc();
+      Frame *item = frame_manager.alloc(file_desc, i);
+      ASSERT_NE(item, nullptr);
+      item->set_file_desc(file_desc);
+      item->set_page_num(i);
       used_list.push_back(item);
     }
 
-    ASSERT_EQ(used_list.size(), frame_manager.get_used_num());
+    ASSERT_EQ(used_list.size(), frame_manager.frame_num());
   }
 }
 
 TEST(test_frame_manager, test_frame_manager_simple_lru)
 {
   BPFrameManager frame_manager("Test");
-  frame_manager.init(false, 2);
+  frame_manager.init(2);
 
   test_get(frame_manager);
 
diff --git a/unitest/bplus_tree_test.cpp b/unitest/bplus_tree_test.cpp
index dabfe08ee6d5815955b97560c94ce56dad5177b6..8bcfd5533a6cde49619f6306d0d1b2fcc0a6f963 100644
--- a/unitest/bplus_tree_test.cpp
+++ b/unitest/bplus_tree_test.cpp
@@ -404,6 +404,12 @@ TEST(test_bplus_tree, test_internal_index_node_handle)
   rid.page_num = 0;
   rid.slot_num = 0;
 
+  key = 0;
+  index = internal_node.lookup(key_comparator, key_mem, &found, &insert_position);
+  ASSERT_EQ(false, found);
+  ASSERT_EQ(0, index);
+  ASSERT_EQ(1, insert_position);
+
   key = 3;
   internal_node.create_new_root(1, key_mem, key);
   for (int i = 2; i < 5; i++) {
diff --git a/unitest/lower_bound_test.cpp b/unitest/lower_bound_test.cpp
new file mode 100644
index 0000000000000000000000000000000000000000..3f5b11852b2062616a28c248ee92581e1ce26e6e
--- /dev/null
+++ b/unitest/lower_bound_test.cpp
@@ -0,0 +1,52 @@
+/* Copyright (c) 2021 Xie Meiyi(xiemeiyi@hust.edu.cn) and OceanBase and/or its affiliates. All rights reserved.
+miniob is licensed under Mulan PSL v2.
+You can use this software according to the terms and conditions of the Mulan PSL v2.
+You may obtain a copy of Mulan PSL v2 at:
+         http://license.coscl.org.cn/MulanPSL2
+THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
+EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
+MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
+See the Mulan PSL v2 for more details. */
+
+//
+// Created by Wangyunlai on 2022/9/5.
+//
+
+#include <vector>
+#include "gtest/gtest.h"
+#include "common/lang/lower_bound.h"
+
+using namespace common;
+
+TEST(lower_bound, test_lower_bound)
+{
+  int items[] = {1, 3, 5, 7, 9};
+  const int count = sizeof(items) / sizeof(items[0]);
+  std::vector<int> v(items, items + count);
+
+  for (int i = 0; i <= 10; i++) {
+    bool found = false;
+    std::vector<int>::iterator retval = lower_bound(v.begin(), v.end(), i, &found);
+    ASSERT_EQ(retval - v.begin(), i / 2);
+    ASSERT_EQ(found, (i % 2 != 0));
+  }
+
+  int empty_items = {};
+  const int empty_count = 0;
+  std::vector<int> empty_v(empty_items, empty_items + empty_count);
+  for (int i = 0; i <= 10; i++) {
+    bool found = false;
+    std::vector<int>::iterator retval = lower_bound(empty_v.begin(), empty_v.end(), i, &found);
+    ASSERT_EQ(retval - empty_v.begin(), 0);
+    ASSERT_EQ(found, false);
+  }
+}
+int main(int argc, char **argv)
+{
+  // 分析gtest程序的命令行参数
+  testing::InitGoogleTest(&argc, argv);
+
+  // 调用RUN_ALL_TESTS()运行所有测试用例
+  // main函数返回RUN_ALL_TESTS()的运行结果
+  return RUN_ALL_TESTS();
+}
diff --git a/unitest/record_manager_test.cpp b/unitest/record_manager_test.cpp
index b6004c4e4f329ba36ac1213a7d34ba2391d7f947..06c77e51b7f087c7dd38492384f40bec2c28d3af 100644
--- a/unitest/record_manager_test.cpp
+++ b/unitest/record_manager_test.cpp
@@ -17,7 +17,7 @@ See the Mulan PSL v2 for more details. */
 
 #include "gtest/gtest.h"
 #include "storage/default/disk_buffer_pool.h"
-#include "storage/common/record_manager.h"
+#include "storage/record/record_manager.h"
 
 using namespace common;