MB-6770 Add the recreate option to KVStore::delVBucket()

VBucket reset operation should delete a vbucket database file
and then recreate the empty database file. As the vbucket reset
operation internally invokes KVStore::delVBucket API, this API
should support the recreate option.

Change-Id: I70f429038a9ed572ddc5cc4e9ad046a257a27f66
Reviewed-on: http://review.couchbase.org/21189
Reviewed-by: Jin Lim <jin@couchbase.com>
Reviewed-by: Michael Wiederhold <mike@couchbase.com>
Tested-by: Karan Kumar <karan@couchbase.com>

src/blackhole-kvstore/blackhole.cc

@@ -49,7 +49,7 @@ void BlackholeKVStore::del(const Item &,
     cb.callback(val);
 }
 
-bool BlackholeKVStore::delVBucket(uint16_t)
+bool BlackholeKVStore::delVBucket(uint16_t, bool)
 {
     return true;
 }

src/blackhole-kvstore/blackhole.hh

@@ -63,7 +63,7 @@ public:
     void del(const Item &itm, uint64_t rowid,
              Callback<int> &cb);
 
-    bool delVBucket(uint16_t vbucket);
+    bool delVBucket(uint16_t vbucket, bool recreate);
 
     vbucket_map_t listPersistedVbuckets(void);
 

src/couch-kvstore/couch-kvstore.cc

@@ -511,7 +511,7 @@ void CouchKVStore::del(const Item &itm,
     }
 }
 
-bool CouchKVStore::delVBucket(uint16_t vbucket)
+bool CouchKVStore::delVBucket(uint16_t vbucket, bool recreate)
 {
     assert(!isReadOnly());
     assert(couchNotifier);
@@ -520,7 +520,17 @@ bool CouchKVStore::delVBucket(uint16_t vbucket)
     couchNotifier->delVBucket(vbucket, cb);
     cb.waitForValue();
 
-    cachedVBStates.erase(vbucket);
+    if (recreate) {
+        vbucket_state vbstate(vbucket_state_dead, 0, 0);
+        vbucket_map_t::iterator it = cachedVBStates.find(vbucket);
+        if (it != cachedVBStates.end()) {
+            vbstate.state = it->second.state;
+        }
+        cachedVBStates[vbucket] = vbstate;
+        resetVBucket(vbucket, vbstate);
+    } else {
+        cachedVBStates.erase(vbucket);
+    }
     updateDbFileMap(vbucket, 1, false);
     return cb.val;
 }
@@ -726,7 +736,7 @@ bool CouchKVStore::snapshotStats(const std::map<std::string, std::string> &stats
     return rv;
 }
 
-bool CouchKVStore::setVBucketState(uint16_t vbucketId, vbucket_state vbstate,
+bool CouchKVStore::setVBucketState(uint16_t vbucketId, vbucket_state &vbstate,
                                    uint32_t vb_change_type, bool newfile,
                                    bool notify)
 {

src/couch-kvstore/couch-kvstore.hh

@@ -326,9 +326,10 @@ public:
      * Delete a given vbucket database instance from the underlying storage system
      *
      * @param vbucket vbucket id
+     * @param recreate true if we need to create an empty vbucket after deletion
      * @return true if the vbucket deletion is completed successfully.
      */
-    bool delVBucket(uint16_t vbucket);
+    bool delVBucket(uint16_t vbucket, bool recreate);
 
     /**
      * Retrieve the list of persisted vbucket states
@@ -475,10 +476,10 @@ protected:
     void loadDB(shared_ptr<Callback<GetValue> > cb, bool keysOnly,
                 std::vector<uint16_t> *vbids,
                 couchstore_docinfos_options options=COUCHSTORE_NO_OPTIONS);
-    bool setVBucketState(uint16_t vbucketId, vbucket_state vbstate,
+    bool setVBucketState(uint16_t vbucketId, vbucket_state &vbstate,
                          uint32_t vb_change_type, bool newfile = false,
                          bool notify = true);
-    bool resetVBucket(uint16_t vbucketId, vbucket_state vbstate) {
+    bool resetVBucket(uint16_t vbucketId, vbucket_state &vbstate) {
         return setVBucketState(vbucketId, vbstate, VB_NO_CHANGE, true, false);
     }
 

src/ep.cc

@@ -249,13 +249,13 @@ class VBucketMemoryDeletionCallback : public DispatcherCallback {
 class VBucketDeletionCallback : public DispatcherCallback {
 public:
     VBucketDeletionCallback(EventuallyPersistentStore *e, RCPtr<VBucket> &vb,
-                            EPStats &st, const void* c = NULL) :
+                            EPStats &st, const void* c = NULL, bool rc = false) :
                             ep(e), vbucket(vb->getId()),
-                            stats(st), cookie(c) {}
+                            stats(st), cookie(c), recreate(rc) {}
 
     bool callback(Dispatcher &, TaskId &) {
         hrtime_t start_time(gethrtime());
-        vbucket_del_result result = ep->completeVBucketDeletion(vbucket);
+        vbucket_del_result result = ep->completeVBucketDeletion(vbucket, recreate);
         if (result == vbucket_del_success || result == vbucket_del_invalid) {
             hrtime_t spent(gethrtime() - start_time);
             hrtime_t wall_time = spent / 1000;
@@ -283,6 +283,7 @@ class VBucketDeletionCallback : public DispatcherCallback {
     uint16_t vbucket;
     EPStats &stats;
     const void* cookie;
+    bool recreate;
 };
 
 EventuallyPersistentStore::EventuallyPersistentStore(EventuallyPersistentEngine &theEngine,
@@ -958,13 +959,13 @@ void EventuallyPersistentStore::scheduleVBSnapshot(const Priority &p) {
 }
 
 vbucket_del_result
-EventuallyPersistentStore::completeVBucketDeletion(uint16_t vbid) {
+EventuallyPersistentStore::completeVBucketDeletion(uint16_t vbid, bool recreate) {
     LockHolder lh(vbsetMutex);
 
     RCPtr<VBucket> vb = vbuckets.getBucket(vbid);
     if (!vb || vb->getState() == vbucket_state_dead || vbuckets.isBucketDeletion(vbid)) {
         lh.unlock();
-        if (rwUnderlying->delVBucket(vbid)) {
+        if (rwUnderlying->delVBucket(vbid, recreate)) {
             vbuckets.setBucketDeletion(vbid, false);
             mutationLog.deleteAll(vbid);
             // This is happening in an independent transaction, so
@@ -982,14 +983,17 @@ EventuallyPersistentStore::completeVBucketDeletion(uint16_t vbid) {
 }
 
 void EventuallyPersistentStore::scheduleVBDeletion(RCPtr<VBucket> &vb,
-                                                   const void* cookie=NULL, double delay=0) {
+                                                   const void* cookie,
+                                                   double delay,
+                                                   bool recreate) {
     shared_ptr<DispatcherCallback> mem_cb(new VBucketMemoryDeletionCallback(this, vb));
     nonIODispatcher->schedule(mem_cb, NULL, Priority::VBMemoryDeletionPriority, delay, false);
 
     if (vbuckets.setBucketDeletion(vb->getId(), true)) {
         shared_ptr<DispatcherCallback> cb(new VBucketDeletionCallback(this, vb,
                                                                       stats,
-                                                                      cookie));
+                                                                      cookie,
+                                                                      recreate));
         dispatcher->schedule(cb,
                              NULL, Priority::VBucketDeletionPriority,
                              delay, false);
@@ -1037,8 +1041,8 @@ bool EventuallyPersistentStore::resetVBucket(uint16_t vbid) {
         RCPtr<VBucket> newvb = vbuckets.getBucket(vbid);
         newvb->checkpointManager.resetTAPCursors(vb->checkpointManager.getTAPCursorNames());
 
-        // Clear all the items from the vbucket kv table on disk.
-        scheduleVBDeletion(vb);
+        // Delete the vbucket database file and recreate the empty file
+        scheduleVBDeletion(vb, NULL, 0, true);
         rv = true;
     }
     return rv;

src/ep.hh

@@ -540,7 +540,7 @@ public:
     /**
      * Perform a fast vbucket deletion.
      */
-    vbucket_del_result completeVBucketDeletion(uint16_t vbid);
+    vbucket_del_result completeVBucketDeletion(uint16_t vbid, bool recreate);
 
     /**
      * Deletes a vbucket
@@ -732,7 +732,9 @@ protected:
 private:
 
     void scheduleVBDeletion(RCPtr<VBucket> &vb,
-                            const void* cookie, double delay);
+                            const void* cookie,
+                            double delay = 0,
+                            bool recreate = false);
 
     RCPtr<VBucket> getVBucket(uint16_t vbid, vbucket_state_t wanted_state);
 

src/kvstore.hh

@@ -30,6 +30,10 @@
 typedef std::pair<int, int64_t> mutation_result;
 
 struct vbucket_state {
+    vbucket_state() { }
+    vbucket_state(vbucket_state_t _state, uint64_t _chkid, uint64_t _maxDelSeqNum) :
+        state(_state), checkpointId(_chkid), maxDeletedSeqno(_maxDelSeqNum) { }
+
     vbucket_state_t state;
     uint64_t checkpointId;
     uint64_t maxDeletedSeqno;
@@ -195,7 +199,7 @@ public:
     /**
      * Delete a given vbucket database.
      */
-    virtual bool delVBucket(uint16_t vbucket) = 0;
+    virtual bool delVBucket(uint16_t vbucket, bool recreate = false) = 0;
 
     /**
      * Get a list of all persisted vbuckets (with their states).