tmp work

db/db_impl.cc

@@ -231,6 +231,14 @@ Status DBImpl::NewDB() {
   return s;
 }
 
+Env* DBImpl::GetEnv() const {
+  return env_;
+}
+
+std::string DBImpl::GetDBName() const {
+  return dbname_;
+}
+
 void DBImpl::MaybeIgnoreError(Status* s) const {
   if (s->ok() || options_.paranoid_checks) {
     // No change needed
@@ -1570,7 +1578,6 @@ Status DBImpl::Get(const ReadOptions& options, const Slice& key,
 
     VlogReader vlogReader(file, &reporter);
     Slice key_value;
-    Slice ret_value;
     char* scratch = new char[encoded_len];
 
     if (vlogReader.ReadValue(offset, encoded_len, &key_value, scratch)) {
@@ -1583,6 +1590,7 @@ Status DBImpl::Get(const ReadOptions& options, const Slice& key,
     }
 
     delete file;
+    file = nullptr;
   }
 
   return s;
@@ -1672,6 +1680,8 @@ Status DBImpl::Write(const WriteOptions& options, WriteBatch* updates) {
 
   if (status.ok() && updates != nullptr) {  // nullptr batch is for compactions
     WriteBatch* write_batch = BuildBatchGroup(&last_writer);
+    WriteBatchInternal::SetSequence(write_batch, last_sequence + 1);
+    last_sequence += WriteBatchInternal::Count(write_batch);
 
     // TODO begin gc中的batch全部都是设置好的。此时是不需要设置的。
     if( !write_batch->IsGarbageColletion() ){
@@ -1685,22 +1695,12 @@ Status DBImpl::Write(const WriteOptions& options, WriteBatch* updates) {
         MaybeScheduleGarbageCollection();
       }
       //SetSequence在write_batch中写入本次的sequence
-      WriteBatchInternal::SetSequence(write_batch, last_sequence + 1);
-      // Count返回write_batch中的key-value个数
       last_sequence += WriteBatchInternal::Count(write_batch);
     }
-    vlog_kv_numbers_ += WriteBatchInternal::Count(write_batch);
     // TODO 这里设置last_sequence  是为了照顾离线回收的时候，在map存在的时候需要调用 ConvertQueue 给回收任务分配sequence。
     // TODO 针对多线程调用put的时候，为了避免给gc回收的时候分配的sequence重叠。
-    versions_->SetLastSequence(last_sequence);
-    // TODO end
-
-
-    WriteBatchInternal::SetSequence(write_batch, last_sequence + 1);
-    last_sequence += WriteBatchInternal::Count(write_batch);
-
-    /* TODO */
     vlog_kv_numbers_ += WriteBatchInternal::Count(write_batch);
+    // TODO end
 
     // Add to log and apply to memtable.  We can release the lock
     // during this phase since &w is currently responsible for logging

db/db_impl.h

@@ -56,7 +56,10 @@ class DBImpl : public DB {
   void GetApproximateSizes(const Range* range, int n, uint64_t* sizes) override;
   void CompactRange(const Slice* begin, const Slice* end) override;
 
-  bool ParseVlogValue(Slice key_value, Slice key, std::string& value, uint64_t val_size);
+  static bool ParseVlogValue(Slice key_value, Slice key, std::string& value, uint64_t val_size);
+
+  Env* GetEnv() const;
+  std::string GetDBName() const;
 
   // Extra methods (for testing) that are not in the public DB interface
 

db/fields.cpp

@@ -1,11 +1,16 @@
 #include <iostream>
 #include <utility>
 
-//#include <fstream>
+#include "db/filename.h"
+
+#include "leveldb/env.h"
 
-#include "fields.h"
 #include "util/coding.h"
+
 #include "dbformat.h"
+#include "fields.h"
+#include "vlog_reader.h"
+#include "db_impl.h"
 
 namespace leveldb {
 
@@ -171,97 +176,23 @@ std::string& Fields::operator[](const std::string& field_name) {
 }
 
 /* 通过若干个字段查询 Key */
-std::vector<std::string> Fields::FindKeysByFields(leveldb::DB* db, const FieldArray& fields) {
-    Fields to_fields = Fields(fields);
-    to_fields.Fields::SortFields();
-    FieldArray search_fields_ = to_fields.fields_;
-
-    std::vector<std::string> find_keys;
-
-    Iterator* it = db->NewIterator(leveldb::ReadOptions());
-    for (it->SeekToFirst(); it->Valid(); it->Next()) {
-
-        std::string iter_key = it->key().ToString();
-        if (std::find(find_keys.begin(), find_keys.end(), iter_key) != find_keys.end()){
-            continue;
-        }
-
-        FieldArray iter_fields_ = Fields::ParseValue(it->value().ToString()).fields_;
-
-        if (iter_fields_ == search_fields_ ||
-            std::includes(iter_fields_.begin(), iter_fields_.end(),
-                          search_fields_.begin(), search_fields_.end())) {
-            find_keys.emplace_back(iter_key);
-        }
-    }
-
-    assert(it->status().ok());
-    delete it;
-
-    return find_keys;
-}
-
-//std::vector<std::string> Fields::FindKeysByFields(leveldb::DB* db, const FieldArray& fields) {
+//std::vector<std::string> Fields::FindKeysByFields(leveldb::DB* db, const FieldArray& fields, const std::string& dbname, Env* env) {
 //    Fields to_fields = Fields(fields);
 //    to_fields.Fields::SortFields();
 //    FieldArray search_fields_ = to_fields.fields_;
 //
 //    std::vector<std::string> find_keys;
-//    std::vector<std::string> deleted_keys;
-//
-//    std::ofstream outfile;
-//    outfile.open("/home/workspace/leveldb_kv/test/log.txt");
-//    outfile.clear();
-//    outfile << "aaaaaaaaaaaaaaaaaaa\n";
 //
 //    Iterator* it = db->NewIterator(leveldb::ReadOptions());
 //    for (it->SeekToFirst(); it->Valid(); it->Next()) {
 //
-//        Slice iter_key_slice = it->key();
-//        outfile << "\niter_key_slice: " + iter_key_slice.ToString() + "\n";
-//        const char* p = iter_key_slice.data();
-//        const char* limit = p + iter_key_slice.size();
-////        const char* limit = p + 5;
-//
-//        uint32_t key_length;
-//        const char* key_ptr = GetVarint32Ptr(p, limit, &key_length);
-//
-//        iter_key_slice = Slice(p, iter_key_slice.size() - 8);
-//
-////        outfile << "\niter_key_slice: " + iter_key_slice.ToString() + "\n";
-//
-//        Slice iter_key_for_parse;
-//        if (!GetLengthPrefixedSlice(&iter_key_slice, &iter_key_for_parse)) {
-//            continue;
-//        }
-//
-//        std::string iter_key = iter_key_for_parse.ToString();
-//
-//        outfile << "\niter_key_str: " + iter_key + "\n";
-//
-//        if (std::find(deleted_keys.begin(), deleted_keys.end(), iter_key) != deleted_keys.end()
-//            || std::find(find_keys.begin(), find_keys.end(), iter_key) != find_keys.end()) {
-//            continue;
-//        }
-//
-//        const uint64_t tag = DecodeFixed64(key_ptr + key_length - 8);
-//        if ((tag & 0xff) == kTypeDeletion) {
-//            deleted_keys.emplace_back(iter_key);
-//            continue;
-//        }
-//
-//        outfile << "\niter_tag_str: " + std::to_string(tag) + "\n";
-//
-//        Slice iter_value_slice = it->value();
-//        Slice iter_value_for_parse;
-//        if (!GetLengthPrefixedSlice(&iter_value_slice, &iter_value_for_parse)) {
+//        std::string iter_key = it->key().ToString();
+//        if (std::find(find_keys.begin(), find_keys.end(), iter_key) != find_keys.end()){
 //            continue;
 //        }
 //
-//        outfile << "\niter_value_str: " + iter_value_for_parse.ToString() + "\n";
+//        FieldArray iter_fields_ = Fields::ParseValue(it->value().ToString()).fields_;
 //
-//        FieldArray iter_fields_ = Fields::ParseValue(iter_value_for_parse.ToString()).fields_;
-//        // FieldArray iter_fields_ = Fields::ParseValue(it->value().ToString()).fields_;
 //        if (iter_fields_ == search_fields_ ||
 //            std::includes(iter_fields_.begin(), iter_fields_.end(),
 //                          search_fields_.begin(), search_fields_.end())) {
@@ -275,4 +206,89 @@ std::vector Fields::FindKeysByFields(leveldb::DB* db, const FieldAr
 //    return find_keys;
 //}
 
+std::vector<std::string> Fields::FindKeysByFields(leveldb::DB* db, const FieldArray& fields, DBImpl* impl) {
+  Fields to_fields = Fields(fields);
+  to_fields.Fields::SortFields();
+  FieldArray search_fields_ = to_fields.fields_;
+
+  std::vector<std::string> find_keys;
+
+  Iterator* it = db->NewIterator(leveldb::ReadOptions());
+  for (it->SeekToFirst(); it->Valid(); it->Next()) {
+
+    std::string iter_key = it->key().ToString();
+    if (std::find(find_keys.begin(), find_keys.end(), iter_key) != find_keys.end()){
+      continue;
+    }
+
+    FieldArray iter_value_ = Fields::ParseValue(it->value().ToString()).fields_;
+
+    if (!iter_value_.empty()) {
+      if (iter_value_ == search_fields_ ||
+          std::includes(iter_value_.begin(), iter_value_.end(),
+                        search_fields_.begin(), search_fields_.end())) {
+        find_keys.emplace_back(iter_key);
+      }
+    } else {
+        uint64_t fid;
+        uint64_t kv_offset;
+        uint64_t val_size;
+        Slice vlog_ptr = it->value();
+        if (!(GetVarint64(&vlog_ptr, &fid)
+            && GetVarint64(&vlog_ptr, &kv_offset)
+            && GetVarint64(&vlog_ptr, &val_size)))  {
+          continue;
+        }
+        uint64_t encoded_len = 1 + VarintLength(it->key().size()) + it->key().size() + VarintLength(val_size) + val_size;
+
+        Env* env = impl->GetEnv();
+        std::string dbname = impl->GetDBName();
+
+        std::string fname = LogFileName(dbname, fid);
+        RandomAccessFile* file;
+
+        Status s = env->NewRandomAccessFile(fname, &file);
+        if (!s.ok()) {
+          continue;
+        }
+        struct VlogReporter : public log::VlogReader::Reporter {
+          Status* status;
+          void Corruption(size_t bytes, const Status& s) override {
+            if (this->status->ok()) *this->status = s;
+          }
+        };
+        VlogReporter reporter;
+        log::VlogReader vlogReader(file, &reporter);
+        Slice key_value;
+        std::string vlog_value;
+        char* scratch = new char[encoded_len];
+
+        if (vlogReader.ReadValue(kv_offset, encoded_len, &key_value, scratch)) {
+          if (!DBImpl::ParseVlogValue(key_value, it->key(), vlog_value, val_size)) {
+            s = Status::Corruption("value in vlog isn't match with given key");
+          }
+        } else {
+          s = Status::Corruption("read vlog error");
+        }
+
+        delete file;
+        file = nullptr;
+
+        iter_value_ = Fields::ParseValue(vlog_value).fields_;
+        if (iter_value_ == search_fields_ ||
+            std::includes(iter_value_.begin(), iter_value_.end(),
+                          search_fields_.begin(), search_fields_.end())) {
+          find_keys.emplace_back(iter_key);
+        }
+    }
+  }
+
+//  assert(it->status().ok());
+  delete it;
+
+  return find_keys;
+}
+
+
+
 } // namespace leveldb

db/fields.h

@@ -1,9 +1,11 @@
 #ifndef LEVELDB_FIELDS_H
 #define LEVELDB_FIELDS_H
 
-#include "vector"
 #include "leveldb/db.h"
 
+#include "db_impl.h"
+#include "vector"
+
 namespace leveldb {
 
 using Field = std::pair<std::string, std::string>;
@@ -58,7 +60,7 @@ class Fields {
         std::string& operator[](const std::string& field_name);
 
         /* 通过若干个字段查询 Key */
-        static std::vector<std::string> FindKeysByFields(leveldb::DB* db, const FieldArray& fields);
+        static std::vector<std::string> FindKeysByFields(leveldb::DB* db, const FieldArray& fields, leveldb::DBImpl* impl);
 
         using iterator = std::vector<Field>::iterator;
         using const_iterator = std::vector<Field>::const_iterator;

test/kv_test.cc

@@ -8,7 +8,7 @@ using namespace leveldb;
 
 constexpr int short_value_size = 4;
 constexpr int long_value_size = 32;
-constexpr int data_size = 32;
+constexpr int data_size = 512;
 
 Status OpenDB(std::string dbName, DB **db) {
   std::string rm_command = "rm -rf " + dbName;
@@ -33,20 +33,6 @@ void InsertData(DB *db, int value_size) {
 
 }
 
- void GetData(DB *db, int size = (1 << 30), int value_size = 0) {
-   ReadOptions readOptions;
-   int key_num = data_size / value_size;
-  
-   // 点查
-   srand(42);
-   for (int i = 0; i < 100; i++) {
-     int key_ = rand() % key_num+1;
-     std::string key = std::to_string(key_);
-     std::string value;
-     db->Get(readOptions, key, &value);
-   }
- }
-
  TEST(TestKV, GetValue) {
      DB *db;
      if(OpenDB("testdb_TestKV_short_value", &db).ok() == false) {
@@ -83,14 +69,12 @@ TEST(TestKV, GetLongValue) {
     Status status;
     int key_num = data_size / long_value_size;
     for (int i = 0; i < key_num; i++) {
-//    for (int i = 0; i < key_num - 1; i++) {
-        // int key_ = rand() % key_num+1;
         std::string key = std::to_string(i);
         std::string value;
         std::string expected_value(long_value_size, 'a');
         status = db->Get(readOptions, key, &value);
-        std::cout << key << std::endl;
-        std::cout << status.ToString() << std::endl;
+//        std::cout << key << std::endl;
+//        std::cout << status.ToString() << std::endl;
         ASSERT_TRUE(status.ok());
         EXPECT_EQ(expected_value, value);
     }

test/value_field_test.cc

@@ -18,7 +18,7 @@ Status OpenDB(const std::string& dbName, DB** db) {
 class FieldsTest : public ::testing::Test {
  protected:
   void SetUp() override {
-    Status s = OpenDB("testdb", &db_);
+    Status s = OpenDB(dbname_ , &db_);
     EXPECT_TRUE(s.ok()) << "Failed to open database: " << s.ToString();
   }
 
@@ -28,6 +28,7 @@ class FieldsTest : public ::testing::Test {
   }
 
   DB* db_ = nullptr; // 数据库实例指针。
+  std::string dbname_ = "testdb"; // 记录数据库路径
 };
 
 // 测试各种构造函数
@@ -225,7 +226,12 @@ TEST_F(FieldsTest, TestBulkInsertSerializeDeleteAndFindKeys) {
 
   // 使用 FindKeysByFields 查找包含特定字段的键
   FieldArray fields_to_find = {{"field2", "value2_"}};
-  std::vector<std::string> found_keys = Fields::FindKeysByFields(db_, fields_to_find);
+
+  Options options;
+  options.create_if_missing = true;
+  DBImpl* impl = new DBImpl(options, dbname_);
+
+  std::vector<std::string> found_keys = Fields::FindKeysByFields(db_, fields_to_find, impl);
 
   // 验证找到的键是否正确
   EXPECT_EQ(found_keys.size(), num_entries - 1) << "Expected " << num_entries - 1 << " keys but found " << found_keys.size();
@@ -237,7 +243,7 @@ TEST_F(FieldsTest, TestBulkInsertSerializeDeleteAndFindKeys) {
 
   // 再次查找，这次没有符合条件的字段
   FieldArray no_match_fields = {{"nonexistent_field", ""}};
-  found_keys = Fields::FindKeysByFields(db_, no_match_fields);
+  found_keys = Fields::FindKeysByFields(db_, no_match_fields, impl);
   EXPECT_TRUE(found_keys.empty()) << "Expected an empty result for non-matching fields.";
 }