RocksDB: Introduce lookup table seqno->key

This is used to allow iterating documents by their sequence number,
acting as a secondary index. By iterating over this column family, one
can use the values (which are document keys) to get the documents from
the main column family.

Change-Id: I0dd3472df23d5a4223b8300d555f79746b4a2c5e
Reviewed-on: http://review.couchbase.org/82745
Reviewed-by: Dave Rigby <[email protected]>
Tested-by: Build Bot <[email protected]>

Author: jameseh96

engines/ep/src/rocksdb-kvstore/rocksdb-kvstore.cc

@@ -42,6 +42,9 @@ RocksDBKVStore::~RocksDBKVStore() {
 
 void RocksDBKVStore::open() {
     rdbOptions.create_if_missing = true;
+    rdbOptions.create_missing_column_families = true;
+
+    seqnoCFOptions.comparator = &vbidSeqnoComparator;
 
     /* Use a listener to set the appropriate engine in the
      * flusher threads RocksDB creates. We need the flusher threads to
@@ -58,8 +61,18 @@ void RocksDBKVStore::open() {
     const std::string dbname =
             dbdir + "/rocksdb." + std::to_string(configuration.getShardId());
 
+    std::vector<rocksdb::ColumnFamilyDescriptor> families{
+            rocksdb::ColumnFamilyDescriptor(rocksdb::kDefaultColumnFamilyName,
+                                            rocksdb::ColumnFamilyOptions()),
+
+            rocksdb::ColumnFamilyDescriptor("vbid_seqno_to_key",
+                                            seqnoCFOptions)};
+
+    std::vector<rocksdb::ColumnFamilyHandle*> handles;
+
     rocksdb::DB* dbPtr;
-    rocksdb::Status s = rocksdb::DB::Open(rdbOptions, dbname, &dbPtr);
+    rocksdb::Status s =
+            rocksdb::DB::Open(rdbOptions, dbname, families, &handles, &dbPtr);
 
     if (s.ok()) {
         db.reset(dbPtr);
@@ -68,10 +81,15 @@ void RocksDBKVStore::open() {
                 "RocksDBKVStore::open: failed to open database '" + dbname +
                 "': " + s.ToString());
     }
+
+    defaultFamilyHandle.reset(handles[0]);
+    seqnoFamilyHandle.reset(handles[1]);
 }
 
 void RocksDBKVStore::close() {
     batch.reset();
+    defaultFamilyHandle.reset();
+    seqnoFamilyHandle.reset();
     db.reset();
 }
 
@@ -136,10 +154,13 @@ void RocksDBKVStore::storeItem(const Item& itm) {
     uint16_t vbid = itm.getVBucketId();
     auto k = mkKeyStr(vbid, itm.getKey());
     auto v = mkValSlice(itm);
+    auto seq = mkSeqnoStr(vbid, itm.getBySeqno());
 
     // TODO RDB: check status.
-    rocksdb::Status s = batch->Put(k, v);
-    cb_assert(s.ok());
+    rocksdb::Status s1 = batch->Put(k, v);
+    rocksdb::Status s2 = batch->Put(seqnoFamilyHandle.get(), seq, k);
+    cb_assert(s1.ok());
+    cb_assert(s2.ok());
 }
 
 GetValue RocksDBKVStore::get(const DocKey& key, uint16_t vb, bool fetchDelete) {
@@ -205,15 +226,27 @@ static bool matches_prefix(rocksdb::Slice s, size_t len, const char* p) {
 
 void RocksDBKVStore::delVBucket(uint16_t vb, uint64_t vb_version) {
     std::lock_guard<std::mutex> lg(writeLock);
-    std::unique_ptr<rocksdb::Iterator> it(
-            db->NewIterator(rocksdb::ReadOptions()));
+    rocksdb::WriteBatch delBatch;
+
     const char* prefix(reinterpret_cast<const char*>(&vb));
     std::string start(prefix, sizeof(vb));
-    rocksdb::WriteBatch delBatch;
-    for (it->Seek(start);
-         it->Valid() && matches_prefix(it->key(), sizeof(vb), prefix);
-         it->Next()) {
-        delBatch.Delete(it->key());
+
+    // We must delete both all the documents for the VB,
+    // and all the vbid:seqno=>vbid:key mappings
+    std::vector<rocksdb::ColumnFamilyHandle*> CFHandles{
+            defaultFamilyHandle.get(), seqnoFamilyHandle.get()};
+
+    // makes use of the fact that keys in both CFs are vbid prefixed
+    // if we move to a db per vb this will just be dropping the whole DB.
+    for (auto* handle : CFHandles) {
+        std::unique_ptr<rocksdb::Iterator> it(
+                db->NewIterator(rocksdb::ReadOptions(), handle));
+
+        for (it->Seek(start);
+             it->Valid() && matches_prefix(it->key(), sizeof(vb), prefix);
+             it->Next()) {
+            delBatch.Delete(handle, it->key());
+        }
     }
     rocksdb::Status s = db->Write(writeOptions, &delBatch);
     cb_assert(s.ok());
@@ -268,6 +301,26 @@ void RocksDBKVStore::grokKeySlice(const rocksdb::Slice& s,
     k->assign(s.data() + sizeof(uint16_t), s.size() - sizeof(uint16_t));
 }
 
+std::string RocksDBKVStore::mkSeqnoStr(uint16_t vbid, int64_t seqno) {
+    size_t size = sizeof(uint16_t) + sizeof(int64_t);
+
+    std::string buffer;
+    buffer.reserve(size);
+
+    buffer.append(reinterpret_cast<char*>(&vbid), sizeof(vbid));
+    buffer.append(reinterpret_cast<char*>(&seqno), sizeof(seqno));
+
+    return buffer;
+}
+
+void RocksDBKVStore::grokSeqnoSlice(const rocksdb::Slice& s,
+                                    uint16_t* vb,
+                                    int64_t* seqno) {
+    assert(s.size() == sizeof(uint16_t) + sizeof(int64_t));
+    std::memcpy(vb, s.data(), sizeof(uint16_t));
+    std::memcpy(seqno, s.data() + sizeof(uint16_t), sizeof(int64_t));
+}
+
 rocksdb::Slice RocksDBKVStore::mkValSlice(const Item& item) {
     // Serialize an Item to the format to write to RocksDB.
     // Using the following layout:

engines/ep/src/rocksdb-kvstore/rocksdb-kvstore.h

@@ -50,6 +50,53 @@ class FlushStartListener : public rocksdb::EventListener {
     EventuallyPersistentEngine* engine;
 };
 
+// Used to order the seqno Column Family to support iterating items by seqno
+class VbidSeqnoComparator : public rocksdb::Comparator {
+public:
+    int Compare(const rocksdb::Slice& a,
+                const rocksdb::Slice& b) const override {
+        const auto vbidA = *reinterpret_cast<const uint16_t*>(a.data());
+        const auto vbidB = *reinterpret_cast<const uint16_t*>(b.data());
+
+        const auto seqnoA =
+                *reinterpret_cast<const int64_t*>(a.data() + sizeof(uint16_t));
+        const auto seqnoB =
+                *reinterpret_cast<const int64_t*>(b.data() + sizeof(uint16_t));
+
+        if (vbidA < vbidB) {
+            return -1;
+        }
+        if (vbidA > vbidB) {
+            return +1;
+        }
+        if (seqnoA < seqnoB) {
+            return -1;
+        }
+        if (seqnoA > seqnoB) {
+            return +1;
+        }
+
+        return 0;
+    }
+
+    const char* Name() const override {
+        return "VbidSeqnoComparator";
+        /* Change this if the comparator implementation is altered
+         This is used to ensure the operator with which the DB was
+         created is the same as the one provided when opening the DB.
+         */
+    }
+    /* Additional functions which must be implemented but aren't required
+     *  to do anything, but could be properly implemented in the future
+     *  if beneficial.
+     */
+    void FindShortestSeparator(std::string*,
+                               const rocksdb::Slice&) const override {
+    }
+    void FindShortSuccessor(std::string*) const override {
+    }
+};
+
 /**
  * A persistence store based on rocksdb.
  */
@@ -256,11 +303,16 @@ class RocksDBKVStore : public KVStore {
      * Direct access to the DB.
      */
     std::unique_ptr<rocksdb::DB> db;
-    char* keyBuffer;
+
+    std::unique_ptr<rocksdb::ColumnFamilyHandle> defaultFamilyHandle;
+    std::unique_ptr<rocksdb::ColumnFamilyHandle> seqnoFamilyHandle;
+    VbidSeqnoComparator vbidSeqnoComparator;
+
     char* valBuffer;
     size_t valSize;
 
     rocksdb::Options rdbOptions;
+    rocksdb::ColumnFamilyOptions seqnoCFOptions;
 
     void open();
 
@@ -270,6 +322,9 @@ class RocksDBKVStore : public KVStore {
     rocksdb::SliceParts mkKeySliceParts(uint16_t, const DocKey& k);
     void grokKeySlice(const rocksdb::Slice&, uint16_t*, std::string*);
 
+    std::string mkSeqnoStr(uint16_t vb, int64_t seqno);
+    void grokSeqnoSlice(const rocksdb::Slice&, uint16_t* vb, int64_t* seqno);
+
     void adjustValBuffer(const size_t);
 
     rocksdb::Slice mkValSlice(const Item& item);