补充delete，一并修改了put逻辑，支持put和delete的合并req

fielddb/field_db.cpp

@@ -82,15 +82,15 @@ Again:
     }
     Request *tail = GetHandleInterval();
     WriteBatch KVBatch,IndexBatch,MetaBatch;
+    std::unordered_set<std::string *> batchKeySet;
     Status status;
     if(!tail->isiCreateReq() && !tail->isiDeleteReq()) {
-        // int debug = tail->type_;
         //表明这一个区间并没有涉及index的创建删除
         {
             //1. 构建各个Batch。构建的过程中要保证索引状态的一致性，需要上锁。
             MutexLock iL(&index_mu);
             for(auto *req_ptr : taskqueue_) {
-                req_ptr->ConstructBatch(KVBatch, IndexBatch, MetaBatch, this);
+                req_ptr->ConstructBatch(KVBatch, IndexBatch, MetaBatch, this, batchKeySet);
                 if(req_ptr == tail) break;
             }
         }
@@ -147,7 +147,7 @@ Status FieldDB::Put(const WriteOptions &options, const Slice &key, const Slice &
     // return kvDB_->Put(options, key, value);
 }
 
-// TODO：需要对是否进行index更新做处理
+// 需要对是否进行index更新做处理
 Status FieldDB::PutFields(const WriteOptions &Options, 
                         const Slice &key, const FieldArray &fields) {
     //这里是为了const和slice-string的转换被迫搞得
@@ -161,14 +161,14 @@ Status FieldDB::PutFields(const WriteOptions &Options,
     // return kvDB_->PutFields(Options, key, fields);  
 }
 
-// todo: 删除有索引的key时indexdb也要同步
+// 删除有索引的key时indexdb也要同步
 Status FieldDB::Delete(const WriteOptions &options, const Slice &key) {
-    //
-    // std::string key_ = key.ToString();
-    // DeleteReq req(&key_,&mutex_);
-    // Status status = HandleRequest(req);
-    // return status;
-    return kvDB_->Delete(options, key);
+    
+    std::string key_ = key.ToString();
+    DeleteReq req(&key_,&mutex_);
+    Status status = HandleRequest(req);
+    return status;
+    // return kvDB_->Delete(options, key);
 }
 // TODO：根据updates里面的东西，要对是否需要更新index进行分别处理
 Status FieldDB::Write(const WriteOptions &options, WriteBatch *updates) {
@@ -220,7 +220,8 @@ Status FieldDB::CreateIndexOnField(const std::string& field_name) {
         return req.s;
     }
     WriteBatch KVBatch,IndexBatch,MetaBatch;
-    req.ConstructBatch(KVBatch, IndexBatch, MetaBatch, this);
+    std::unordered_set<std::string *> useless;
+    req.ConstructBatch(KVBatch, IndexBatch, MetaBatch, this, useless);
     indexDB_->Write(WriteOptions(), &IndexBatch);
     req.Finalize(this);
     return req.s;
@@ -235,7 +236,8 @@ Status FieldDB::DeleteIndex(const std::string &field_name) {
         return req.s;
     }
     WriteBatch KVBatch,IndexBatch,MetaBatch;
-    req.ConstructBatch(KVBatch, IndexBatch, MetaBatch, this);
+    std::unordered_set<std::string *> useless;
+    req.ConstructBatch(KVBatch, IndexBatch, MetaBatch, this, useless);
     indexDB_->Write(WriteOptions(), &IndexBatch);
     req.Finalize(this);
     return req.s;

fielddb/meta.cpp

@@ -21,8 +21,8 @@ using namespace leveldb;
 // }
 
 
-//对于含有index field的put的meta编码为 (KV|Key,Value)
-void MetaKV::Trans(Slice &MetaKey,Slice &MetaValue) {
+//对于含有index field的put/delete的meta编码为 (KV|Key,Value)
+void MetaKV::TransPut(Slice &MetaKey,Slice &MetaValue) {
     MetaKey.clear();
     MetaValue.clear();
     std::string buf;
@@ -32,6 +32,14 @@ void MetaKV::Trans(Slice &MetaKey,Slice &MetaValue) {
     MetaValue = Slice(*value);
 }
 
+void MetaKV::TransDelete(Slice &MetaKey) {
+    MetaKey.clear();
+    std::string buf;
+    PutFixed32(&buf, KV_Deleting);
+    PutLengthPrefixedSlice(&buf, Slice(*name));
+    MetaKey = Slice(buf);
+}
+
 class CleanerHandler : public WriteBatch::Handler {
 public:
     WriteBatch *NeedClean;

fielddb/meta.h

@@ -27,7 +27,7 @@ using namespace leveldb;
 // };
 
 enum MetaType {
-    Index, //记录index状态的meta
+    //Index, //记录index状态的meta
     KV_Creating,  //记录含有index field的put的meta
     KV_Deleting,
 };
@@ -35,9 +35,10 @@ enum MetaType {
 //将一对(field_name,field_value)转换到metaDB中的KV表示
 class MetaKV {
 public:
-    MetaKV(std::string *field_name,std::string *field_value):
+    MetaKV(std::string *field_name,std::string *field_value = nullptr):
         name(field_name),value(field_value) { }
-    void Trans(Slice &MetaKey,Slice &MetaValue);
+    void TransPut(Slice &MetaKey,Slice &MetaValue);
+    void TransDelete(Slice &MetaKey);
 private:
     std::string *name;
     std::string *value;

fielddb/request.cpp

@@ -19,7 +19,7 @@ void Request::PendReq(Request *req) {
 
 //为虚函数提供最基本的实现
 void Request::ConstructBatch(WriteBatch &KVBatch,WriteBatch &IndexBatch,
-                                    WriteBatch &MetaBatch,fielddb::FieldDB *DB)
+                WriteBatch &MetaBatch,fielddb::FieldDB *DB,std::unordered_set<std::string *> &batchKeySet)
 {
     assert(0);
 }
@@ -41,10 +41,28 @@ bool Request::isPending() {
 
 /*******FieldsReq*******/
 void FieldsReq::ConstructBatch(WriteBatch &KVBatch,WriteBatch &IndexBatch,
-                                    WriteBatch &MetaBatch,fielddb::FieldDB *DB)
+                    WriteBatch &MetaBatch,fielddb::FieldDB *DB,std::unordered_set<std::string *> &batchKeySet)
 {
+    if (batchKeySet.find(Key) != batchKeySet.end()){
+        return;//并发的被合并的put/delete请求只处理一次
+    } else {
+        batchKeySet.insert(Key);
+    }
+    std::string val_str;
+    Status s = DB->kvDB_->Get(ReadOptions(), *Key, &val_str);
+    FieldArray *oldFields;
+    if (s.IsNotFound()){
+        oldFields = nullptr;
+    } else if (s.ok()) { //得到数据库之前key的fields, 判断需不需要删除其中潜在的索引
+        oldFields = ParseValue(val_str);        
+    } else {
+        assert(0);
+    }
+
+
     KVBatch.Put(Slice(*Key), Slice(SerializeValue(*Fields)));
     bool HasIndex = false;
+    bool HasOldIndex = false;
     {
         // MutexLock L(&DB->index_mu); //互斥访问索引状态表
         DB->index_mu.AssertHeld();
@@ -62,39 +80,118 @@ void FieldsReq::ConstructBatch(WriteBatch &KVBatch,WriteBatch &IndexBatch,
                 //assert(0);
             }
         }
+        //冲突也可能存在于，需要删除旧数据的索引，但该索引正在创删中
+        if (oldFields != nullptr){
+            for(auto [field_name,field_value] : *oldFields) {
+                if(field_name == "") break;
+                if(DB->index_.count(field_name)) {
+                    auto [index_status,parent_req] = DB->index_[field_name];
+                    if(index_status == IndexStatus::Creating || index_status == IndexStatus::Deleting) {
+                        parent_req->PendReq(this);
+                        return;
+                    } else if(index_status == IndexStatus::Exist) {
+                        HasOldIndex = true;
+                    }
+                    //assert(0);
+                }
+            }
+        }
+
         //2.对于没有冲突但含有索引操作的put，构建metaKV，这里直接将KV对简单编码后写入metaDB
-        if(HasIndex) {
+        if(HasIndex || HasOldIndex) {
             Slice MetaKey,MetaValue;
             std::string serialized = SerializeValue(*Fields);
             MetaKV MKV = MetaKV(Key,&serialized);
-            MKV.Trans(MetaKey, MetaValue);
+            MKV.TransPut(MetaKey, MetaValue);
             MetaBatch.Put(MetaKey, MetaValue);
-            //第三点是不是应该在这一分支中
-        //3.对于含有索引的field建立索引
-            for(auto [field_name,field_value] : *Fields) {
-                if(field_name == "") continue;
-                if(DB->index_.count(field_name)) {
-                    std::string indexKey;
-                    AppendIndexKey(&indexKey, ParsedInternalIndexKey(
-                                                *Key,field_name,field_value));
-                    IndexBatch.Put(indexKey, Slice());
+            
+        
+        //3.1对于含有索引的oldfield删除索引
+            if (HasOldIndex) {
+                for(auto [field_name,field_value] : *oldFields) {
+                    if(field_name == "") continue;
+                    if(DB->index_.count(field_name)) {
+                        std::string indexKey;
+                        AppendIndexKey(&indexKey, ParsedInternalIndexKey(
+                                                    *Key,field_name,field_value));
+                        IndexBatch.Delete(indexKey);
+                    }
                 }
             }
+
+        //3.2对于含有索引的field建立索引
+            if (HasIndex) {
+                for(auto [field_name,field_value] : *Fields) {
+                    if(field_name == "") continue;
+                    if(DB->index_.count(field_name)) {
+                        std::string indexKey;
+                        AppendIndexKey(&indexKey, ParsedInternalIndexKey(
+                                                    *Key,field_name,field_value));
+                        IndexBatch.Put(indexKey, Slice());
+                    }
+                }
+            }
+            
         }
+        //优化：对于3.1，3.2中都有的索引只写一次
     }
 }
 
 
 /*******DeleteReq*******/
 void DeleteReq::ConstructBatch(WriteBatch &KVBatch,WriteBatch &IndexBatch,
-        WriteBatch &MetaBatch,fielddb::FieldDB *DB)
+        WriteBatch &MetaBatch,fielddb::FieldDB *DB,std::unordered_set<std::string *> &batchKeySet)
 {
-    //TODO:
+    if (batchKeySet.find(Key) != batchKeySet.end()){
+        return;//并发的被合并的put/delete请求只处理一次
+    } else {
+        batchKeySet.insert(Key);
+    }
     //1. 读取当前的最新的键值对，判断是否存在含有键值对的field
     //2.1 如果无，则正常构造delete
     //2.2 如果是有的field的索引状态都是exist，则在meta中写KV_Deleting类型的记录
     //在kvDB和metaDB中写入对应的delete
     //2.3 如果存在field的索引状态是Creating或者Deleting，那么在那个队列上面进行等待
+    std::string val_str;
+    DB->kvDB_->Get(ReadOptions(), *Key, &val_str);
+    FieldArray *Fields = ParseValue(val_str);
+    KVBatch.Delete(Slice(*Key));
+    bool HasIndex = false;
+    {
+        // MutexLock L(&DB->index_mu); //互斥访问索引状态表
+        DB->index_mu.AssertHeld();
+        //1.将存在冲突的delete pend到对应的请求
+        for(auto [field_name,field_value] : *Fields) {
+            if(field_name == "") break;
+            if(DB->index_.count(field_name)) {
+                auto [index_status,parent_req] = DB->index_[field_name];
+                if(index_status == IndexStatus::Creating || index_status == IndexStatus::Deleting) {
+                    parent_req->PendReq(this);
+                    return;
+                } else if(index_status == IndexStatus::Exist) {
+                    HasIndex = true;
+                }
+                //assert(0);
+            }
+        }
+        //2.对于没有冲突但含有索引操作的delete，构建metaKV，这里直接将KV对简单编码后写入metaDB
+        if(HasIndex) {
+            Slice MetaKey;
+            MetaKV MKV = MetaKV(Key);
+            MKV.TransDelete(MetaKey); //meta中写入一个delete不需要value
+            MetaBatch.Put(MetaKey, Slice());
+        //3.对于含有索引的field删除索引
+            for(auto [field_name,field_value] : *Fields) {
+                if(field_name == "") continue;
+                if(DB->index_.count(field_name)) {
+                    std::string indexKey;
+                    AppendIndexKey(&indexKey, ParsedInternalIndexKey(
+                                                *Key,field_name,field_value));
+                    IndexBatch.Delete(indexKey);
+                }
+            }
+        }
+    }
 }
 
 /*******iCreateReq*******/
@@ -126,7 +223,7 @@ void iCreateReq::PendReq(Request *req) {
 }
 
 void iCreateReq::ConstructBatch(WriteBatch &KVBatch,WriteBatch &IndexBatch,
-        WriteBatch &MetaBatch,fielddb::FieldDB *DB)
+        WriteBatch &MetaBatch,fielddb::FieldDB *DB,std::unordered_set<std::string *> &batchKeySet)
 {
     //遍历数据库，构建二级索引到indexbatch，(更新metaDB中的元数据为Index类型的（Field,Creating）)
     //一个indexwritebatch写入，那么索引创建删除应该和metadb没有交互
@@ -185,7 +282,7 @@ void iDeleteReq::PendReq(Request* req) {
 }
 
 void iDeleteReq::ConstructBatch(WriteBatch &KVBatch,WriteBatch &IndexBatch,
-        WriteBatch &MetaBatch,fielddb::FieldDB *DB)
+        WriteBatch &MetaBatch,fielddb::FieldDB *DB,std::unordered_set<std::string *> &batchKeySet)
 {
     std::vector<std::pair<std::string, std::string>> keysAndVal = 
                                             DB->FindKeysAndValByFieldName(*Field);

fielddb/request.h

@@ -5,6 +5,7 @@
 #include "port/port_stdcxx.h"
 #include "util/mutexlock.h"
 #include "util/serialize_value.h"
+#include <unordered_set>
 // #include "fielddb/field_db.h"
 
 #ifndef REQUEST_H
@@ -43,7 +44,7 @@ public:
 
     //用于含有Fields的
     virtual void ConstructBatch(WriteBatch &KVBatch,WriteBatch &IndexBatch,
-                WriteBatch &MetaBatch,fielddb::FieldDB *DB);
+                WriteBatch &MetaBatch,fielddb::FieldDB *DB,std::unordered_set<std::string *> &batchKeySet);
     //主要用于icreate和idelete在队列中的注册当前状态
     virtual void Prepare(FieldDB *DB);
     virtual void Finalize(FieldDB *DB);
@@ -65,7 +66,7 @@ public:
         Key(Key),Fields(Fields),Request(FieldsReq_t,mu) { };
 
     void ConstructBatch(WriteBatch &KVBatch,WriteBatch &IndexBatch,
-        WriteBatch &MetaBatch,fielddb::FieldDB *DB) override;
+        WriteBatch &MetaBatch,fielddb::FieldDB *DB,std::unordered_set<std::string *> &batchKeySet) override;
 
     std::string *Key;
     FieldArray *Fields;
@@ -90,7 +91,7 @@ public:
         Field(Field),Request(iCreateReq_t, mu),Existed(false) { };
 
     void ConstructBatch(WriteBatch &KVBatch,WriteBatch &IndexBatch,
-        WriteBatch &MetaBatch,fielddb::FieldDB *DB) override;
+        WriteBatch &MetaBatch,fielddb::FieldDB *DB,std::unordered_set<std::string *> &batchKeySet) override;
     void Prepare(FieldDB *DB) override;
     void Finalize(FieldDB *DB) override;
     void PendReq(Request *req) override;
@@ -107,7 +108,7 @@ public:
         Field(Field),Request(iDeleteReq_t, mu),Deleted(false) { };
 
     void ConstructBatch(WriteBatch &KVBatch,WriteBatch &IndexBatch,
-        WriteBatch &MetaBatch,fielddb::FieldDB *DB) override;
+        WriteBatch &MetaBatch,fielddb::FieldDB *DB,std::unordered_set<std::string *> &batchKeySet) override;
     void Prepare(FieldDB *DB) override;
     void Finalize(FieldDB *DB) override;
     void PendReq(Request *req) override;
@@ -124,7 +125,7 @@ public:
         Key(Key),Request(DeleteReq_t,mu) { };
 
     void ConstructBatch(WriteBatch &KVBatch,WriteBatch &IndexBatch,
-        WriteBatch &MetaBatch,fielddb::FieldDB *DB) override;
+        WriteBatch &MetaBatch,fielddb::FieldDB *DB,std::unordered_set<std::string *> &batchKeySet) override;
 
     std::string *Key;
 };

test/basic_function_test.cc

@@ -19,6 +19,10 @@ TEST(TestLab1, Basic) {
     bool allowNotFound = false;
     GetFieldData(db, allowNotFound);    
     findKeysByCity(db);
+
+    DeleteFieldData(db);
+    GetDeleteData(db);
+
     delete db;
 }
 
@@ -40,10 +44,19 @@ TEST(TestLab2, Basic) {
     db->CreateIndexOnField("age");
     findKeysByCityIndex(db, true);
     findKeysByAgeIndex(db, true);
+
     db->DeleteIndex("address");
     findKeysByCityIndex(db, false);
     findKeysByAgeIndex(db, true);
 
+    DeleteFieldData(db);
+    // GetDeleteData(db);
+    //helper太长不再封装函数了，这里因为数据都被delete了，但索引还在，所以能QueryByIndex但返回key数量0
+    Field field = {"age", "20"};
+    Status s;
+    std::vector<std::string> resKeys = db->QueryByIndex(field, &s);
+    ASSERT_EQ(resKeys.size(), 0);
+
     delete db;
 }
 

test/helper.cc

@@ -15,7 +15,9 @@ std::vector cities = {
 //检查insert和queryByIndex的数据是否对应
 std::set<std::string> shanghaiKeys; 
 std::set<std::string> age20Keys;
-//复杂的测试要注意这两个全局变量，目前只有InsertFieldData和InsertOneField会往里加，并且没有清理
+//复杂的测试要注意这两个全局变量，
+//目前只有InsertFieldData和InsertOneField会往里加，DeleteFieldData会全部清空，
+//其他测试之间有必要手动clear
 
 Status OpenDB(std::string dbName, FieldDB **db) {
   Options options;
@@ -98,6 +100,25 @@ void InsertFieldData(FieldDB *db, int seed = 0/*随机种子*/) {
   }
 }
 
+void DeleteFieldData(FieldDB *db, int seed = 0/*随机种子*/) {
+  std::cout << "-------deleting-------" << std::endl;
+  WriteOptions writeOptions;
+  int key_num = data_size / value_size;
+  // srand线程不安全，这种可以保证多线程时随机序列也一致
+  std::mt19937 rng(seed);
+
+  shanghaiKeys.clear();
+  age20Keys.clear();
+  for (int i = 0; i < key_num; i++) {
+    int randThisTime = rng(); //确保读写一个循环只rand一次，否则随机序列会不一致
+    int key_ = std::abs(randThisTime) % key_num + 1; 
+    std::string key = std::to_string(key_);
+
+    Status s =  db->Delete(WriteOptions(), key);
+    ASSERT_TRUE(s.ok());
+  }
+}
+
 //并发时不一定能读到，加个参数控制
 void GetFieldData(FieldDB *db, bool allowNotFound, int seed = 0) {
   std::cout << "-------getting-------" << std::endl;
@@ -136,6 +157,24 @@ void GetFieldData(FieldDB *db, bool allowNotFound, int seed = 0) {
   }
 }
 
+//检查对应种子有没有删除干净
+//删除期间即使其他种子也不能并发写，因为即使种子不同，随机出的key可能相同
+void GetDeleteData(FieldDB *db, int seed = 0) {
+  std::cout << "-------getting-------" << std::endl;
+  ReadOptions readOptions;
+  int key_num = data_size / value_size;
+  
+  std::mt19937 rng(seed);
+  for (int i = 0; i < 100; i++) {
+    int randThisTime = rng();
+    int key_ = std::abs(randThisTime) % key_num + 1;
+    std::string key = std::to_string(key_);
+    FieldArray fields_ret;
+    Status s = db->GetFields(readOptions, key, &fields_ret);
+    ASSERT_TRUE(s.IsNotFound());
+  }
+}
+
 void findKeysByCity(FieldDB *db) {
   std::cout << "-------getting field address-------" << std::endl;
   Field field = {"address", "Shanghai"};