From 041bede9e4ac643601d16deb01a530748bc7d99e Mon Sep 17 00:00:00 2001
From: VirgilZhu <94546750@qq.com>
Date: Thu, 12 Dec 2024 03:56:38 +0800
Subject: [PATCH 1/4] kv sep v0.1: add vlog_manager/reader/writer interface

---
 CMakeLists.txt            |  5 +++-
 db/vlog_manager.h         | 68 +++++++++++++++++++++++++++++++++++++++++++++++
 db/vlog_reader.h          | 35 ++++++++++++++++++++++++
 db/vlog_writer.h          | 39 +++++++++++++++++++++++++++
 include/leveldb/options.h |  6 +++++
 5 files changed, 152 insertions(+), 1 deletion(-)
 create mode 100644 db/vlog_manager.h
 create mode 100644 db/vlog_reader.h
 create mode 100644 db/vlog_writer.h

diff --git a/CMakeLists.txt b/CMakeLists.txt
index 974f190..d210d34 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -117,7 +117,10 @@ endif(BUILD_SHARED_LIBS)
 # Must be included before CMAKE_INSTALL_INCLUDEDIR is used.
 include(GNUInstallDirs)
 
-add_library(leveldb)
+add_library(leveldb
+        db/vlog_reader.h
+        db/vlog_writer.h
+        db/vlog_manager.h)
 target_sources(leveldb
   PRIVATE
     "${PROJECT_BINARY_DIR}/${LEVELDB_PORT_CONFIG_DIR}/port_config.h"
diff --git a/db/vlog_manager.h b/db/vlog_manager.h
new file mode 100644
index 0000000..07d14d3
--- /dev/null
+++ b/db/vlog_manager.h
@@ -0,0 +1,68 @@
+#ifndef LEVELDB_VLOG_MANAGER_H
+#define LEVELDB_VLOG_MANAGER_H
+
+#include <map>
+#include <set>
+#include <cstdint>
+#include "leveldb/env.h"
+
+namespace leveldb {
+namespace vlog {
+
+class VlogWriter;
+class VlogReader;
+
+class VlogInfo {
+    private:
+        /* vlog 当前大小 */
+        size_t size_;
+        /* vlog 当前若插入新 record 的偏移 */
+        size_t head_;
+        /* vlog 当前过期的 record 数量 */
+        uint64_t expired_count_;
+
+        VlogReader* vlog_read_;
+        VlogWriter* vlog_write_;
+
+    public:
+        VlogInfo() : size_(0), head_(0), expired_count_(0) {}
+        ~VlogInfo() = default;
+
+    friend class VlogWriter;
+    friend class VlogReader;
+    friend class VlogManager;
+};
+
+class VlogManager {
+    public:
+        explicit VlogManager(uint64_t expired_threshold);
+        ~VlogManager() = default;
+
+        /* 新建一个 vlog 并编号 */
+        void AddVlog(uint64_t vlog_no);
+        /* 往当前活跃 vlog 插入一条 record */
+        Status AddRecord(const Slice& slice);
+        /* 更新当前活跃 vlog 的头部位置 */
+        Status SetHead(size_t offset);
+        /* 当前活跃 vlog 的写缓冲的数据全部写入磁盘 */
+        Status Sync();
+        /* 从某个 vlog 中获取 value */
+        Status FetchValueFromVlog(Slice record_addr, std::string* value);
+        /* 设置当前活跃 vlog */
+        void SetCurrentVlog(uint64_t vlog_no);
+
+    private:
+        /* 管理所有 vlog 的编号和信息 */
+        std::map<uint64_t, VlogInfo*> vlog_manager_;
+        /* 记录正在 GC 的 vlog 集合 */
+        std::set<uint64_t> gc_vlog_set_;
+        /* 记录触发 GC 的过期阈值 */
+        uint64_t expired_threshold_;
+        /* 记录当前活跃的 vlog 编号 */
+        uint64_t cur_vlog_;
+};
+
+} // namespace vlog
+} // namespace leveldb
+
+#endif  // LEVELDB_VLOG_MANAGER_H
diff --git a/db/vlog_reader.h b/db/vlog_reader.h
new file mode 100644
index 0000000..b4c2baa
--- /dev/null
+++ b/db/vlog_reader.h
@@ -0,0 +1,35 @@
+#ifndef LEVELDB_VLOG_READER_H
+#define LEVELDB_VLOG_READER_H
+
+#include <cstdint>
+#include "leveldb/slice.h"
+#include "leveldb/status.h"
+
+namespace leveldb {
+
+class RandomAccessFile;
+
+namespace vlog {
+
+class VlogInfo;
+class VlogManager;
+
+class VlogReader {
+    public:
+        VlogReader(uint32_t vlog_no);
+        ~VlogReader() = default;
+
+        /* 在 vlog_offset 处往后读取一条 record 的 value */
+        Status Get(uint64_t vlog_offset, std::string* value);
+
+    private:
+        VlogInfo* vlog_info_;
+        RandomAccessFile* vlog_;
+
+    friend class VlogManager;
+};
+
+} // namespace vlog
+} // namespace leveldb
+
+#endif  // LEVELDB_VLOG_READER_H
diff --git a/db/vlog_writer.h b/db/vlog_writer.h
new file mode 100644
index 0000000..afc07d7
--- /dev/null
+++ b/db/vlog_writer.h
@@ -0,0 +1,39 @@
+#ifndef LEVELDB_VLOG_WRITER_H
+#define LEVELDB_VLOG_WRITER_H
+
+#include <map>
+#include "leveldb/status.h"
+
+namespace leveldb {
+
+class WritableFile;
+
+namespace vlog {
+
+class VlogInfo;
+class VlogManager;
+
+class VlogWriter {
+    public:
+        explicit VlogWriter(WritableFile* vlog);
+        ~VlogWriter();
+
+        /* 往 vlog 的写缓存写入一条 kv 记录 */
+        Status AddRecord(const Slice& slice);
+
+    private:
+        VlogWriter() = default;
+
+        VlogInfo* vlog_info_;
+        WritableFile* vlog_;
+
+        VlogWriter(const VlogWriter&);
+        VlogWriter& operator=(const VlogWriter&);
+
+    friend class VlogManager;
+};
+
+}  // namespace vlog
+}  // namespace leveldb
+
+#endif  // LEVELDB_VLOG_WRITER_H
diff --git a/include/leveldb/options.h b/include/leveldb/options.h
index d755f46..d87c112 100644
--- a/include/leveldb/options.h
+++ b/include/leveldb/options.h
@@ -145,6 +145,12 @@ struct LEVELDB_EXPORT Options {
   // Many applications will benefit from passing the result of
   // NewBloomFilterPolicy() here.
   const FilterPolicy* filter_policy = nullptr;
+
+  // vlog 过期 kv 计数器触发 GC 的阈值
+  int expired_threshold;
+
+  // vlog 文件大小上限
+  int max_vlog_size = 64 * 1024 * 1024;
 };
 
 // Options that control read operations

From 1d7f56bbd8ec0292c497343721e7465c6898061d Mon Sep 17 00:00:00 2001
From: GUJIEJASON <1776127334@qq.com>
Date: Thu, 12 Dec 2024 12:10:18 +0800
Subject: [PATCH 2/4] Update README

---
 README.md | 13 ++++++++++---
 1 file changed, 10 insertions(+), 3 deletions(-)

diff --git a/README.md b/README.md
index 0692880..e15fab2 100755
--- a/README.md
+++ b/README.md
@@ -33,8 +33,15 @@
   3. 允许在存储 KV 键值对前修改字段内容；
   4. 允许通过字段值查询所有对应的 key。
 + **实现思路：**
-  1. 设计类 `Fields` 来操作字段数组，与字段相关的字段截取、读写操作、序列化等函数均在 `Fields` 类中实现；
-  2. 通过字段查询 Key ：实现函数 `FindKeysByField`，传入若干字段名和字段值（即子字段数组），遍历查找 LSM-Tree 找到对应的若干 key。
+  1. 定义一个` Fields`类来管理`LevelDB`中的字段。其中`Field`是使用标准库中的`std::pair<std::string, std::string>` 定义了单个字段的格式，而`FieldArray`则是使用了`std::vector<Field>` 来定义一组字段。
+  2. 定义`fields_`这一私有成员变量，它是一个 `FieldArray` 类型的向量，用来存储一组字段。
+  3. 定义一系列的构造函数来支持从不同类型的参数创建`Fields`对象。
+  4. 定义了`SortFields`方法来确保在创建`Fields`对象时，各个字段会根据`field_name`从小到大进行排序，进而减少后续更新删除操作中会出现的通过 `field_name` 遍历 `Fields` 的耗时。
+  5. 定义了`UpdateField` 和 `UpdateFields` 方法允许用户更新或插入单个或多个字段，以及`DeleteField` 和 `DeleteFields` 方法允许用户删除单个或多个字段。实现思路是通过遍历fields_来查找匹配的field_name并对其进行更新（若不存在则插入）以及删除操作（⭐在上述操作的实现中，由于字段列表的有序性，遍历时可通过比较field_name大小来提前判断该字段是否存在，对于小字段的查找尤其明显，进而有效减少搜索时间，提高搜索效率。）
+  6. 定义`SerializeValue` 和 `ParseValue` 方法分别用于将字段序列化为字符串或将字符串反序列化为字段对象。实现时是调用了`conding.h`文件中的`PutLengthPrefixedSlice`函数以及`GetLengthPrefixedSlice`函数，它们的作用分别是对一个`string`进行编码并在其前面加入长度信息和将编码后的`string`中的长度信息去除，提取出原始`string`。这两个函数不仅可以完美实现我们对字段编码的最初设计，同时也和`lsm-tree`里原来所有的`kv`数据对的编码保持一致。
+  7. 定义`GetField`和`HasField`方法用于访问特定字段和检查字段是否存在，实现思路与更新删除操作类似，也是对`fields_`进行遍历。
+  8. 重载 `[]` 运算符，以提供类似字典的字段访问方式。对于常量对象，`operator[]` 返回字段值的副本，如果给定的字段名不存在，则返回一个空字符串，并输出错误信息。而对于非常量对象，`operator[]` 返回字段值的引用，并允许修改该字段值。如果给定的字段名不存在，则会插入一个新的字段，并返回新字段值的引用。
+  9. 定义了一个静态方法`FindKeysByFields` 用于根据若干个字段在数据库中查找对应的键。实现上是使用`LevelDB`提供的`API`创建一个`NewIterator`,从数据库的第一个条目遍历到最后一个条目。事先定义了`find_keys` 来存储找到的键，在遍历过程中，为了避免重复处理同一个键，会先检查当前键是否已经存在于 `find_keys` 中。如果存在，则跳过此条目。若不存在，则提取其`value`部分，利用 `ParseValue` 方法将字符串形式的值解析为 `Fields` 对象，进而获得该条目对应的字段数组。再对解析后的字段数组与`search_fields_`进行匹配，这里支持完全匹配和部分匹配，将匹配的`key`存入`find_keys`中，最后返回`find_keys`。(💡Tips：在使用`Iterator`进行遍历时，`it.key()`和`it.value()`获取的其实是`kv`字符串本身，不需要我们再解码`kv length`和考虑`tag`（`ktypevalue`、`ktypedeletion`）。我们在设计之初并未考虑到这一点，是在后续测试的`debug`中发现了这一情况)
 
 ### 2.2  KV 分离
 
@@ -75,7 +82,7 @@
     class Fields {
         private:
             FieldArray fields_;
-
+  
         public:
             /* 从 FieldArray 构造 */
             explicit Fields(const FieldArray& fields);

From 04e46dde2753a2c8d964993179c31df8151d7e59 Mon Sep 17 00:00:00 2001
From: GUJIEJASON <1776127334@qq.com>
Date: Fri, 27 Dec 2024 16:15:39 +0800
Subject: [PATCH 3/4] kv start

---
 db/vlog_manager.h | 68 -------------------------------------------------------
 db/vlog_reader.h  | 35 ----------------------------
 db/vlog_writer.h  | 39 -------------------------------
 3 files changed, 142 deletions(-)
 delete mode 100644 db/vlog_manager.h
 delete mode 100644 db/vlog_reader.h
 delete mode 100644 db/vlog_writer.h

diff --git a/db/vlog_manager.h b/db/vlog_manager.h
deleted file mode 100644
index 07d14d3..0000000
--- a/db/vlog_manager.h
+++ /dev/null
@@ -1,68 +0,0 @@
-#ifndef LEVELDB_VLOG_MANAGER_H
-#define LEVELDB_VLOG_MANAGER_H
-
-#include <map>
-#include <set>
-#include <cstdint>
-#include "leveldb/env.h"
-
-namespace leveldb {
-namespace vlog {
-
-class VlogWriter;
-class VlogReader;
-
-class VlogInfo {
-    private:
-        /* vlog 当前大小 */
-        size_t size_;
-        /* vlog 当前若插入新 record 的偏移 */
-        size_t head_;
-        /* vlog 当前过期的 record 数量 */
-        uint64_t expired_count_;
-
-        VlogReader* vlog_read_;
-        VlogWriter* vlog_write_;
-
-    public:
-        VlogInfo() : size_(0), head_(0), expired_count_(0) {}
-        ~VlogInfo() = default;
-
-    friend class VlogWriter;
-    friend class VlogReader;
-    friend class VlogManager;
-};
-
-class VlogManager {
-    public:
-        explicit VlogManager(uint64_t expired_threshold);
-        ~VlogManager() = default;
-
-        /* 新建一个 vlog 并编号 */
-        void AddVlog(uint64_t vlog_no);
-        /* 往当前活跃 vlog 插入一条 record */
-        Status AddRecord(const Slice& slice);
-        /* 更新当前活跃 vlog 的头部位置 */
-        Status SetHead(size_t offset);
-        /* 当前活跃 vlog 的写缓冲的数据全部写入磁盘 */
-        Status Sync();
-        /* 从某个 vlog 中获取 value */
-        Status FetchValueFromVlog(Slice record_addr, std::string* value);
-        /* 设置当前活跃 vlog */
-        void SetCurrentVlog(uint64_t vlog_no);
-
-    private:
-        /* 管理所有 vlog 的编号和信息 */
-        std::map<uint64_t, VlogInfo*> vlog_manager_;
-        /* 记录正在 GC 的 vlog 集合 */
-        std::set<uint64_t> gc_vlog_set_;
-        /* 记录触发 GC 的过期阈值 */
-        uint64_t expired_threshold_;
-        /* 记录当前活跃的 vlog 编号 */
-        uint64_t cur_vlog_;
-};
-
-} // namespace vlog
-} // namespace leveldb
-
-#endif  // LEVELDB_VLOG_MANAGER_H
diff --git a/db/vlog_reader.h b/db/vlog_reader.h
deleted file mode 100644
index b4c2baa..0000000
--- a/db/vlog_reader.h
+++ /dev/null
@@ -1,35 +0,0 @@
-#ifndef LEVELDB_VLOG_READER_H
-#define LEVELDB_VLOG_READER_H
-
-#include <cstdint>
-#include "leveldb/slice.h"
-#include "leveldb/status.h"
-
-namespace leveldb {
-
-class RandomAccessFile;
-
-namespace vlog {
-
-class VlogInfo;
-class VlogManager;
-
-class VlogReader {
-    public:
-        VlogReader(uint32_t vlog_no);
-        ~VlogReader() = default;
-
-        /* 在 vlog_offset 处往后读取一条 record 的 value */
-        Status Get(uint64_t vlog_offset, std::string* value);
-
-    private:
-        VlogInfo* vlog_info_;
-        RandomAccessFile* vlog_;
-
-    friend class VlogManager;
-};
-
-} // namespace vlog
-} // namespace leveldb
-
-#endif  // LEVELDB_VLOG_READER_H
diff --git a/db/vlog_writer.h b/db/vlog_writer.h
deleted file mode 100644
index afc07d7..0000000
--- a/db/vlog_writer.h
+++ /dev/null
@@ -1,39 +0,0 @@
-#ifndef LEVELDB_VLOG_WRITER_H
-#define LEVELDB_VLOG_WRITER_H
-
-#include <map>
-#include "leveldb/status.h"
-
-namespace leveldb {
-
-class WritableFile;
-
-namespace vlog {
-
-class VlogInfo;
-class VlogManager;
-
-class VlogWriter {
-    public:
-        explicit VlogWriter(WritableFile* vlog);
-        ~VlogWriter();
-
-        /* 往 vlog 的写缓存写入一条 kv 记录 */
-        Status AddRecord(const Slice& slice);
-
-    private:
-        VlogWriter() = default;
-
-        VlogInfo* vlog_info_;
-        WritableFile* vlog_;
-
-        VlogWriter(const VlogWriter&);
-        VlogWriter& operator=(const VlogWriter&);
-
-    friend class VlogManager;
-};
-
-}  // namespace vlog
-}  // namespace leveldb
-
-#endif  // LEVELDB_VLOG_WRITER_H

From af6de4df42e1cf4ee8e3c85874bc25a25a6fe05c Mon Sep 17 00:00:00 2001
From: GUJIEJASON <1776127334@qq.com>
Date: Fri, 27 Dec 2024 19:05:07 +0800
Subject: [PATCH 4/4] Add VlogWriter

---
 db/db_impl.cc                 |  8 +++--
 db/db_impl.h                  |  3 ++
 db/vlog_writer.cc             | 45 ++++++++++++++++++++++++++
 db/vlog_writer.h              | 44 ++++++++++++++++++++++++++
 db/write_batch.cc             | 73 +++++++++++++++++++++++++++++++++++++++++++
 db/write_batch_internal.h     |  2 ++
 include/leveldb/write_batch.h |  1 +
 7 files changed, 174 insertions(+), 2 deletions(-)
 create mode 100644 db/vlog_writer.cc
 create mode 100644 db/vlog_writer.h

diff --git a/db/db_impl.cc b/db/db_impl.cc
index 3a5c225..06f8e61 100644
--- a/db/db_impl.cc
+++ b/db/db_impl.cc
@@ -1249,7 +1249,11 @@ Status DBImpl::Write(const WriteOptions& options, WriteBatch* updates) {
     // into mem_.
     {
       mutex_.Unlock();
-      status = log_->AddRecord(WriteBatchInternal::Contents(write_batch));
+      // 先写入vlog再写入memtable
+      // 写vlog日志 offset 表示这个 write_batch 在vlog中的偏移地址。
+      uint64_t offset = 0;
+      status = vlog_->AddRecord(WriteBatchInternal::Contents(write_batch),offset);
+      // status = log_->AddRecord(WriteBatchInternal::Contents(write_batch));
       bool sync_error = false;
       if (status.ok() && options.sync) {
         status = logfile_->Sync();
@@ -1258,7 +1262,7 @@ Status DBImpl::Write(const WriteOptions& options, WriteBatch* updates) {
         }
       }
       if (status.ok()) {
-        status = WriteBatchInternal::InsertInto(write_batch, mem_);
+        status = WriteBatchInternal::InsertInto(write_batch, mem_, logfile_number_, offset);
       }
       mutex_.Lock();
       if (sync_error) {
diff --git a/db/db_impl.h b/db/db_impl.h
index 91f4fd4..81db57d 100644
--- a/db/db_impl.h
+++ b/db/db_impl.h
@@ -12,6 +12,7 @@
 
 #include "db/dbformat.h"
 #include "db/log_writer.h"
+#include "db/vlog_writer.h"
 #include "db/snapshot.h"
 #include "leveldb/db.h"
 #include "leveldb/env.h"
@@ -207,6 +208,8 @@ class DBImpl : public DB {
   Status bg_error_ GUARDED_BY(mutex_);
 
   CompactionStats stats_[config::kNumLevels] GUARDED_BY(mutex_);
+
+  log::VlogWriter* vlog_;
 };
 
 // Sanitize db options.  The caller should delete result.info_log if
diff --git a/db/vlog_writer.cc b/db/vlog_writer.cc
new file mode 100644
index 0000000..b07e2bc
--- /dev/null
+++ b/db/vlog_writer.cc
@@ -0,0 +1,45 @@
+#include "db/vlog_writer.h"
+
+#include <cstdint>
+
+#include "leveldb/env.h"
+
+#include "util/coding.h"
+#include "util/crc32c.h"
+
+namespace leveldb {
+namespace log {
+VlogWriter::VlogWriter(WritableFile* dest) : dest_(dest), head_(0) {}
+
+VlogWriter::VlogWriter(WritableFile* dest, uint64_t dest_length)
+    : dest_(dest), head_(0) {}
+
+VlogWriter::~VlogWriter() = default;
+
+Status VlogWriter::AddRecord(const Slice& slice, uint64_t& offset) {
+  const char* ptr = slice.data();
+  size_t left = slice.size();
+  Status s;
+  s = EmitPhysicalRecord(ptr, left, offset);
+  return s;
+}
+
+Status VlogWriter::EmitPhysicalRecord(const char* ptr, size_t length,
+                                      uint64_t& offset) {
+  assert(length <= 0xffff);
+  char buf[4];
+  EncodeFixed32(buf, length);
+  Status s = dest_->Append(Slice(buf, 4));
+  if (s.ok()) {
+    s = dest_->Append(Slice(ptr, length));
+    if (s.ok()) {
+      s = dest_->Flush();
+      offset = head_ + 4;
+      head_ += 4 + length;
+    }
+  }
+  return s;
+}
+
+}  // namespace log
+}  // namespace leveldb
diff --git a/db/vlog_writer.h b/db/vlog_writer.h
new file mode 100644
index 0000000..9cb2313
--- /dev/null
+++ b/db/vlog_writer.h
@@ -0,0 +1,44 @@
+#ifndef LEVELDB_DB_VLOG_WRITER_H_
+#define LEVELDB_DB_VLOG_WRITER_H_
+
+#include "db/log_format.h"
+#include <cstdint>
+
+#include "leveldb/slice.h"
+#include "leveldb/status.h"
+
+namespace leveldb {
+
+class WritableFile;
+
+namespace log {
+
+class VlogWriter {
+ public:
+  // Create a writer that will append data to "*dest".
+  // "*dest" must be initially empty.
+  // "*dest" must remain live while this Writer is in use.
+  explicit VlogWriter(WritableFile* dest);
+
+  // Create a writer that will append data to "*dest".
+  // "*dest" must have initial length "dest_length".
+  // "*dest" must remain live while this Writer is in use.
+  VlogWriter(WritableFile* dest, uint64_t dest_length);
+
+  VlogWriter(const VlogWriter&) = delete;
+  VlogWriter& operator=(const VlogWriter&) = delete;
+
+  VlogWriter();
+
+  Status AddRecord(const Slice& slice, uint64_t& offset);
+
+ private:
+  Status EmitPhysicalRecord(const char* ptr, size_t length, uint64_t& offset);
+  size_t head_;
+  WritableFile* dest_;
+};
+
+}  // namespace log
+}  // namespace leveldb
+
+#endif  // LEVELDB_DB_VLOG_WRITER_H_
diff --git a/db/write_batch.cc b/db/write_batch.cc
index b54313c..874bc34 100644
--- a/db/write_batch.cc
+++ b/db/write_batch.cc
@@ -18,7 +18,9 @@
 #include "db/dbformat.h"
 #include "db/memtable.h"
 #include "db/write_batch_internal.h"
+
 #include "leveldb/db.h"
+
 #include "util/coding.h"
 
 namespace leveldb {
@@ -79,6 +81,68 @@ Status WriteBatch::Iterate(Handler* handler) const {
   }
 }
 
+Status WriteBatch::Iterate(Handler* handler, uint64_t fid,
+                           uint64_t offset) const {
+  Slice input(rep_);
+  // 整个writebatch 的起始地址
+  const char* begin = input.data();
+
+  if (input.size() < kHeader) {
+    return Status::Corruption("malformed WriteBatch (too small)");
+  }
+  // 12个字节，8个字节用来表示sequence，4个字节用来表示 count，移除头
+  input.remove_prefix(kHeader);
+  Slice key, value;
+  int found = 0;
+  while (!input.empty()) {
+    found++;
+    const uint64_t kv_offset = input.data() - begin + offset;
+    assert(kv_offset > 0);
+
+    // record 记录为  1 个字节 是 kTypeValue ,剩下的字节是 key value
+    // record 记录为  1 个字节 是 kTypeDeletion， 剩下的字节是key
+    char tag = input[0];
+    input.remove_prefix(1);
+    switch (tag) {
+      case kTypeValue:
+        if (GetLengthPrefixedSlice(&input, &key) &&
+            GetLengthPrefixedSlice(&input, &value)) {
+          handler->Put(key, value);
+        } else {
+          return Status::Corruption("bad WriteBatch Put");
+        }
+        break;
+      case kTypeDeletion:
+        if (GetLengthPrefixedSlice(&input, &key)) {
+          handler->Delete(key);
+        } else {
+          return Status::Corruption("bad WriteBatch Delete");
+        }
+        break;
+      // case kTypeSeparate:
+      //   if (GetLengthPrefixedSlice(&input, &key) &&
+      //       GetLengthPrefixedSlice(&input, &(value))) {
+      //     // value = fileNumber + offset + valuesize 采用变长编码的方式
+      //     std::string dest;
+      //     PutVarint64(&dest, fid);
+      //     PutVarint64(&dest, kv_offset);
+      //     PutVarint64(&dest, value.size());
+      //     Slice value_offset(dest);
+      //     handler->Put(key, value_offset, kTypeSeparate);
+      //   } else {
+      //     return Status::Corruption("bad WriteBatch Put");
+      //   }
+      //   break;
+      default:
+        return Status::Corruption("unknown WriteBatch tag");
+    }
+  }
+  if (found != WriteBatchInternal::Count(this)) {
+    return Status::Corruption("WriteBatch has wrong count");
+  } else {
+    return Status::OK();
+  }
+}
 int WriteBatchInternal::Count(const WriteBatch* b) {
   return DecodeFixed32(b->rep_.data() + 8);
 }
@@ -136,6 +200,15 @@ Status WriteBatchInternal::InsertInto(const WriteBatch* b, MemTable* memtable) {
   return b->Iterate(&inserter);
 }
 
+Status WriteBatchInternal::InsertInto(const WriteBatch* b, MemTable* memtable,
+                                      uint64_t fid, size_t offset) {
+  MemTableInserter inserter;
+  // 一批 writeBatch中只有一个sequence，公用的，后续会自加。
+  inserter.sequence_ = WriteBatchInternal::Sequence(b);
+  inserter.mem_ = memtable;
+  return b->Iterate(&inserter, fid, offset);
+}
+
 void WriteBatchInternal::SetContents(WriteBatch* b, const Slice& contents) {
   assert(contents.size() >= kHeader);
   b->rep_.assign(contents.data(), contents.size());
diff --git a/db/write_batch_internal.h b/db/write_batch_internal.h
index fce86e3..dd08be8 100644
--- a/db/write_batch_internal.h
+++ b/db/write_batch_internal.h
@@ -37,6 +37,8 @@ class WriteBatchInternal {
 
   static Status InsertInto(const WriteBatch* batch, MemTable* memtable);
 
+  static Status InsertInto(const WriteBatch* batch, MemTable* memtable,uint64_t fid, size_t offset);
+
   static void Append(WriteBatch* dst, const WriteBatch* src);
 };
 
diff --git a/include/leveldb/write_batch.h b/include/leveldb/write_batch.h
index 94d4115..2af98a3 100644
--- a/include/leveldb/write_batch.h
+++ b/include/leveldb/write_batch.h
@@ -71,6 +71,7 @@ class LEVELDB_EXPORT WriteBatch {
 
   // Support for iterating over the contents of a batch.
   Status Iterate(Handler* handler) const;
+  Status Iterate(Handler* handler, uint64_t fid, uint64_t offset) const;
 
  private:
   friend class WriteBatchInternal;