Remove removePrefix interface in kvstore

src/kvstore/KVEngine.h

@@ -23,8 +23,6 @@ class WriteBatch {
 
     virtual ResultCode remove(folly::StringPiece key) = 0;
 
-    virtual ResultCode removePrefix(folly::StringPiece prefix) = 0;
-
     // Remove all keys in the range [start, end)
     virtual ResultCode removeRange(folly::StringPiece start,
                                    folly::StringPiece end) = 0;
@@ -44,7 +42,9 @@ class KVEngine {
     virtual const char* getDataRoot() const = 0;
 
     virtual std::unique_ptr<WriteBatch> startBatchWrite() = 0;
-    virtual ResultCode commitBatchWrite(std::unique_ptr<WriteBatch> batch) = 0;
+
+    virtual ResultCode commitBatchWrite(std::unique_ptr<WriteBatch> batch,
+                                        bool disableWAL = true) = 0;
 
     // Read a single key
     virtual ResultCode get(const std::string& key, std::string* value) = 0;
@@ -84,9 +84,6 @@ class KVEngine {
     virtual ResultCode removeRange(const std::string& start,
                                    const std::string& end) = 0;
 
-    // Remove rows with the given prefix
-    virtual ResultCode removePrefix(const std::string& prefix) = 0;
-
     // Add partId into current storage engine.
     virtual void addPart(PartitionID partId) = 0;
 

src/kvstore/KVStore.h

@@ -153,11 +153,6 @@ class KVStore {
                                   const std::string& end,
                                   KVCallback cb) = 0;
 
-    virtual void asyncRemovePrefix(GraphSpaceID spaceId,
-                                   PartitionID partId,
-                                   const std::string& prefix,
-                                   KVCallback cb) = 0;
-
     virtual void asyncAtomicOp(GraphSpaceID spaceId,
                                PartitionID partId,
                                raftex::AtomicOp op,

src/kvstore/LogEncoder.h

@@ -17,7 +17,6 @@ enum LogType : char {
     OP_MULTI_PUT      = 0x2,
     OP_REMOVE         = 0x3,
     OP_MULTI_REMOVE   = 0x4,
-    OP_REMOVE_PREFIX  = 0x5,
     OP_REMOVE_RANGE   = 0x6,
     OP_ADD_LEARNER    = 0x07,
     OP_TRANS_LEADER   = 0x08,
@@ -30,7 +29,6 @@ enum BatchLogType : char {
     OP_BATCH_PUT            = 0x1,
     OP_BATCH_REMOVE         = 0x2,
     OP_BATCH_REMOVE_RANGE   = 0x3,
-    OP_BATCH_REMOVE_PREFIX  = 0x4,
 };
 
 std::string encodeKV(const folly::StringPiece& key,
@@ -86,13 +84,6 @@ class BatchHolder {
         batch_.emplace_back(std::move(op));
     }
 
-    void removePrefix(std::string&& key) {
-        auto op = std::make_tuple(BatchLogType::OP_BATCH_REMOVE_PREFIX,
-                                  std::forward<std::string>(key),
-                                  "");
-        batch_.emplace_back(std::move(op));
-    }
-
     void clear() {
         batch_.clear();
     }

src/kvstore/NebulaStore.cpp

@@ -520,20 +520,6 @@ void NebulaStore::asyncRemoveRange(GraphSpaceID spaceId,
     part->asyncRemoveRange(start, end, std::move(cb));
 }
 
-
-void NebulaStore::asyncRemovePrefix(GraphSpaceID spaceId,
-                                    PartitionID partId,
-                                    const std::string& prefix,
-                                    KVCallback cb) {
-    auto ret = part(spaceId, partId);
-    if (!ok(ret)) {
-        cb(error(ret));
-        return;
-    }
-    auto part = nebula::value(ret);
-    part->asyncRemovePrefix(prefix, std::move(cb));
-}
-
 void NebulaStore::asyncAtomicOp(GraphSpaceID spaceId,
                                 PartitionID partId,
                                 raftex::AtomicOp op,

src/kvstore/NebulaStore.h

@@ -171,11 +171,6 @@ class NebulaStore : public KVStore, public Handler {
                           const std::string& end,
                           KVCallback cb) override;
 
-    void asyncRemovePrefix(GraphSpaceID spaceId,
-                           PartitionID partId,
-                           const std::string& prefix,
-                           KVCallback cb) override;
-
     void asyncAtomicOp(GraphSpaceID spaceId,
                        PartitionID partId,
                        raftex::AtomicOp op,

src/kvstore/Part.cpp

@@ -7,6 +7,7 @@
 #include "kvstore/Part.h"
 #include "kvstore/LogEncoder.h"
 #include "base/NebulaKeyUtils.h"
+#include "kvstore/RocksEngineConfig.h"
 
 DEFINE_int32(cluster_id, 0, "A unique id for each cluster");
 
@@ -98,16 +99,6 @@ void Part::asyncMultiRemove(const std::vector<std::string>& keys, KVCallback cb)
 }
 
 
-void Part::asyncRemovePrefix(folly::StringPiece prefix, KVCallback cb) {
-    std::string log = encodeSingleValue(OP_REMOVE_PREFIX, prefix);
-
-    appendAsync(FLAGS_cluster_id, std::move(log))
-        .thenValue([this, callback = std::move(cb)] (AppendLogResult res) mutable {
-            callback(this->toResultCode(res));
-        });
-}
-
-
 void Part::asyncRemoveRange(folly::StringPiece start,
                             folly::StringPiece end,
                             KVCallback cb) {
@@ -249,14 +240,6 @@ bool Part::commitLogs(std::unique_ptr<LogIterator> iter) {
             }
             break;
         }
-        case OP_REMOVE_PREFIX: {
-            auto prefix = decodeSingleValue(log);
-            if (batch->removePrefix(prefix) != ResultCode::SUCCEEDED) {
-                LOG(ERROR) << idStr_ << "Failed to call WriteBatch::removePrefix()";
-                return false;
-            }
-            break;
-        }
         case OP_REMOVE_RANGE: {
             auto range = decodeMultiValues(log);
             DCHECK_EQ(2, range.size());
@@ -276,10 +259,7 @@ bool Part::commitLogs(std::unique_ptr<LogIterator> iter) {
                     code = batch->remove(op.second.first);
                 } else if (op.first == BatchLogType::OP_BATCH_REMOVE_RANGE) {
                     code = batch->removeRange(op.second.first, op.second.second);
-                } else if (op.first == BatchLogType::OP_BATCH_REMOVE_PREFIX) {
-                    code = batch->removePrefix(op.second.first);
                 }
-
                 if (code != ResultCode::SUCCEEDED) {
                     LOG(ERROR) << idStr_ << "Failed to call WriteBatch";
                     return false;
@@ -325,7 +305,8 @@ bool Part::commitLogs(std::unique_ptr<LogIterator> iter) {
             return false;
         }
     }
-    return engine_->commitBatchWrite(std::move(batch)) == ResultCode::SUCCEEDED;
+    return engine_->commitBatchWrite(std::move(batch), FLAGS_rocksdb_disable_wal)
+                    == ResultCode::SUCCEEDED;
 }
 
 std::pair<int64_t, int64_t> Part::commitSnapshot(const std::vector<std::string>& rows,
@@ -350,7 +331,8 @@ std::pair<int64_t, int64_t> Part::commitSnapshot(const std::vector<std::string>&
             return std::make_pair(0, 0);
         }
     }
-    if (ResultCode::SUCCEEDED != engine_->commitBatchWrite(std::move(batch))) {
+    // For snapshot, we open the rocksdb's wal to avoid loss data if crash.
+    if (ResultCode::SUCCEEDED != engine_->commitBatchWrite(std::move(batch), false)) {
         LOG(ERROR) << idStr_ << "Put failed in commit";
         return std::make_pair(0, 0);
     }

src/kvstore/Part.h

@@ -45,7 +45,6 @@ class Part : public raftex::RaftPart {
 
     void asyncRemove(folly::StringPiece key, KVCallback cb);
     void asyncMultiRemove(const std::vector<std::string>& keys, KVCallback cb);
-    void asyncRemovePrefix(folly::StringPiece prefix, KVCallback cb);
     void asyncRemoveRange(folly::StringPiece start,
                           folly::StringPiece end,
                           KVCallback cb);

src/kvstore/RocksEngine.cpp

@@ -27,10 +27,9 @@ namespace {
 class RocksWriteBatch : public WriteBatch {
 private:
     rocksdb::WriteBatch batch_;
-    rocksdb::DB* db_{nullptr};
 
 public:
-    explicit RocksWriteBatch(rocksdb::DB* db) : batch_(FLAGS_rocksdb_batch_size), db_(db) {}
+    RocksWriteBatch() : batch_(FLAGS_rocksdb_batch_size) {}
 
     virtual ~RocksWriteBatch() = default;
 
@@ -50,25 +49,6 @@ class RocksWriteBatch : public WriteBatch {
         }
     }
 
-    ResultCode removePrefix(folly::StringPiece prefix) override {
-        rocksdb::Slice pre(prefix.begin(), prefix.size());
-        rocksdb::ReadOptions options;
-        std::unique_ptr<rocksdb::Iterator> iter(db_->NewIterator(options));
-        iter->Seek(pre);
-        while (iter->Valid()) {
-            if (iter->key().starts_with(pre)) {
-                if (!batch_.Delete(iter->key()).ok()) {
-                    return ResultCode::ERR_UNKNOWN;
-                }
-            } else {
-                // Done
-                break;
-            }
-            iter->Next();
-        }
-        return ResultCode::SUCCEEDED;
-    }
-
     // Remove all keys in the range [start, end)
     ResultCode removeRange(folly::StringPiece start, folly::StringPiece end) override {
         if (batch_.DeleteRange(toSlice(start), toSlice(end)).ok()) {
@@ -127,13 +107,13 @@ RocksEngine::RocksEngine(GraphSpaceID spaceId,
 
 
 std::unique_ptr<WriteBatch> RocksEngine::startBatchWrite() {
-    return std::make_unique<RocksWriteBatch>(db_.get());
+    return std::make_unique<RocksWriteBatch>();
 }
 
 
-ResultCode RocksEngine::commitBatchWrite(std::unique_ptr<WriteBatch> batch) {
+ResultCode RocksEngine::commitBatchWrite(std::unique_ptr<WriteBatch> batch, bool disableWAL) {
     rocksdb::WriteOptions options;
-    options.disableWAL = FLAGS_rocksdb_disable_wal;
+    options.disableWAL = disableWAL;
     auto* b = static_cast<RocksWriteBatch*>(batch.get());
     rocksdb::Status status = db_->Write(options, b->data());
     if (status.ok()) {
@@ -285,8 +265,6 @@ ResultCode RocksEngine::removeRange(const std::string& start,
                                     const std::string& end) {
     rocksdb::WriteOptions options;
     options.disableWAL = FLAGS_rocksdb_disable_wal;
-    // TODO(sye) Given the RocksDB version we are using,
-    // we should avoud using DeleteRange
     auto status = db_->DeleteRange(options, db_->DefaultColumnFamily(), start, end);
     if (status.ok()) {
         return ResultCode::SUCCEEDED;
@@ -296,41 +274,10 @@ ResultCode RocksEngine::removeRange(const std::string& start,
     }
 }
 
-
-ResultCode RocksEngine::removePrefix(const std::string& prefix) {
-    rocksdb::Slice pre(prefix.data(), prefix.size());
-    rocksdb::ReadOptions readOptions;
-    rocksdb::WriteBatch batch;
-    std::unique_ptr<rocksdb::Iterator> iter(db_->NewIterator(readOptions));
-    iter->Seek(pre);
-    while (iter->Valid()) {
-        if (iter->key().starts_with(pre)) {
-            auto status = batch.Delete(iter->key());
-            if (!status.ok()) {
-                return ResultCode::ERR_UNKNOWN;
-            }
-        } else {
-            // Done
-            break;
-        }
-        iter->Next();
-    }
-
-    rocksdb::WriteOptions writeOptions;
-    writeOptions.disableWAL = FLAGS_rocksdb_disable_wal;
-    if (db_->Write(writeOptions, &batch).ok()) {
-        return ResultCode::SUCCEEDED;
-    } else {
-        return ResultCode::ERR_UNKNOWN;
-    }
-}
-
-
 std::string RocksEngine::partKey(PartitionID partId) {
     return NebulaKeyUtils::systemPartKey(partId);
 }
 
-
 void RocksEngine::addPart(PartitionID partId) {
     auto ret = put(partKey(partId), "");
     if (ret == ResultCode::SUCCEEDED) {

src/kvstore/RocksEngine.h

@@ -108,7 +108,9 @@ class RocksEngine : public KVEngine {
     }
 
     std::unique_ptr<WriteBatch> startBatchWrite() override;
-    ResultCode commitBatchWrite(std::unique_ptr<WriteBatch> batch) override;
+
+    ResultCode commitBatchWrite(std::unique_ptr<WriteBatch> batch,
+                                bool disableWAL) override;
 
     /*********************
      * Data retrieval
@@ -143,8 +145,6 @@ class RocksEngine : public KVEngine {
     ResultCode removeRange(const std::string& start,
                            const std::string& end) override;
 
-    ResultCode removePrefix(const std::string& prefix) override;
-
     /*********************
      * Non-data operation
      ********************/

src/kvstore/plugins/hbase/HBaseStore.h

@@ -156,7 +156,7 @@ class HBaseStore : public KVStore {
     void asyncRemovePrefix(GraphSpaceID spaceId,
                            PartitionID partId,
                            const std::string& prefix,
-                           KVCallback cb) override;
+                           KVCallback cb);
 
     void asyncAtomicOp(GraphSpaceID,
                        PartitionID,

src/storage/BaseProcessor.h

@@ -68,8 +68,6 @@ class BaseProcessor {
     void doRemoveRange(GraphSpaceID spaceId, PartitionID partId, std::string start,
                        std::string end);
 
-    void doRemovePrefix(GraphSpaceID spaceId, PartitionID partId, std::string prefix);
-
     kvstore::ResultCode doRange(GraphSpaceID spaceId, PartitionID partId, std::string start,
                                 std::string end, std::unique_ptr<kvstore::KVIterator>* iter);
 

src/storage/BaseProcessor.inl

@@ -104,16 +104,6 @@ void BaseProcessor<RESP>::doRemoveRange(GraphSpaceID spaceId,
         });
 }
 
-template <typename RESP>
-void BaseProcessor<RESP>::doRemovePrefix(GraphSpaceID spaceId,
-                                         PartitionID partId,
-                                         std::string prefix) {
-    this->kvstore_->asyncRemovePrefix(
-        spaceId, partId, prefix, [spaceId, partId, this](kvstore::ResultCode code) {
-            handleAsync(spaceId, partId, code);
-        });
-}
-
 template<typename RESP>
 kvstore::ResultCode BaseProcessor<RESP>::doRange(GraphSpaceID spaceId,
                                                  PartitionID partId,

src/storage/mutate/DeleteVerticesProcessor.cpp

@@ -30,29 +30,31 @@ void DeleteVerticesProcessor::process(const cpp2::DeleteVerticesRequest& req) {
             const auto& vertices = pv->second;
             for (auto v = vertices.begin(); v != vertices.end(); v++) {
                 auto prefix = NebulaKeyUtils::vertexPrefix(part, *v);
+                std::unique_ptr<kvstore::KVIterator> iter;
+                auto ret = this->kvstore_->prefix(spaceId, part, prefix, &iter);
+                if (ret != kvstore::ResultCode::SUCCEEDED) {
+                    VLOG(3) << "Error! ret = " << static_cast<int32_t>(ret)
+                            << ", spaceID " << spaceId;
+                    this->onFinished();
+                    return;
+                }
 
-                // Evict vertices from cache
-                if (FLAGS_enable_vertex_cache && vertexCache_ != nullptr) {
-                    std::unique_ptr<kvstore::KVIterator> iter;
-                    auto ret = this->kvstore_->prefix(spaceId, part, prefix, &iter);
-                    if (ret != kvstore::ResultCode::SUCCEEDED) {
-                        VLOG(3) << "Error! ret = " << static_cast<int32_t>(ret)
-                                << ", spaceID " << spaceId;
-                        this->onFinished();
-                        return;
-                    }
-
-                    while (iter->valid()) {
-                        auto key = iter->key();
-                        if (NebulaKeyUtils::isVertex(key)) {
-                            auto tag = NebulaKeyUtils::getTagId(key);
+                std::vector<std::string> keys;
+                keys.reserve(32);
+                while (iter->valid()) {
+                    auto key = iter->key();
+                    if (NebulaKeyUtils::isVertex(key)) {
+                        auto tag = NebulaKeyUtils::getTagId(key);
+                        // Evict vertices from cache
+                        if (FLAGS_enable_vertex_cache && vertexCache_ != nullptr) {
                             VLOG(3) << "Evict vertex cache for VID " << *v << ", TagID " << tag;
                             vertexCache_->evict(std::make_pair(*v, tag), part);
                         }
-                        iter->next();
+                        keys.emplace_back(key.str());
                     }
+                    iter->next();
                 }
-                doRemovePrefix(spaceId, part, std::move(prefix));
+                doRemove(spaceId, part, std::move(keys));
             }
         }
     } else {