RocksDB: Implement RocksDBKVStore::scan basics

Lays the initial groundwork for iterating all keys by seqno.

At the moment, the ColumnFamily mapping seqno=>key will grow forever; we
do not ever remove old mappings. This could be done in a compaction
filter, or a periodic task etc. The main costs currently are slower
scans as time is wasted ignoring seqno=>key mappings where the seqno is
no longer current, and ever growing storage usage.

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

Author: jameseh96

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

@@ -656,16 +656,133 @@ ScanContext* RocksDBKVStore::initScanContext(
         uint64_t startSeqno,
         DocumentFilter options,
         ValueFilter valOptions) {
-    // TODO RDB vmx 2016-10-29: implement
     size_t scanId = scanCounter++;
+    // As we cannot efficiently determine how many documents this scan will
+    // find, we approximate this value with the seqno difference + 1
+    // as scan is supposed to be inclusive at both ends,
+    // seqnos 2 to 4 covers 3 docs not 4 - 2 = 2
+
+    uint64_t endSeqno = cachedVBStates[vbid]->highSeqno;
     return new ScanContext(cb,
                            cl,
                            vbid,
                            scanId,
                            startSeqno,
-                           99999999,
+                           endSeqno,
                            options,
                            valOptions,
-                           999999,
+                           /* documentCount */ endSeqno - startSeqno + 1,
                            configuration);
 }
+
+scan_error_t RocksDBKVStore::scan(ScanContext* ctx) {
+    if (!ctx) {
+        return scan_failed;
+    }
+
+    if (ctx->lastReadSeqno == ctx->maxSeqno) {
+        return scan_success;
+    }
+
+    uint64_t start = ctx->startSeqno;
+    if (ctx->lastReadSeqno != 0) {
+        start = ctx->lastReadSeqno + 1;
+    }
+
+    GetMetaOnly isMetaOnly = ctx->valFilter == ValueFilter::KEYS_ONLY
+                                     ? GetMetaOnly::Yes
+                                     : GetMetaOnly::No;
+
+    rocksdb::ReadOptions snapshotOpts{rocksdb::ReadOptions()};
+    snapshotOpts.snapshot = db->GetSnapshot();
+
+    std::unique_ptr<rocksdb::Iterator> it(
+            db->NewIterator(snapshotOpts, seqnoFamilyHandle.get()));
+
+    std::string startStr = mkSeqnoStr(ctx->vbid, start);
+    it->Seek(startStr);
+
+    std::string endStr = mkSeqnoStr(ctx->vbid, ctx->maxSeqno);
+
+    auto isPastEnd = [&endStr, this](rocksdb::Slice seqSlice) {
+        return vbidSeqnoComparator.Compare(seqSlice, endStr) == 1;
+    };
+
+    for (; it->Valid() && !isPastEnd(it->key()); it->Next()) {
+        uint16_t vb;
+        int64_t seqno;
+
+        rocksdb::Slice seqnoSlice = it->key();
+        grokSeqnoSlice(seqnoSlice, &vb, &seqno);
+
+        rocksdb::Slice keySlice = it->value();
+
+        std::string valueStr;
+        rocksdb::Status s = db->Get(snapshotOpts, keySlice, &valueStr);
+
+        if (!s.ok()) {
+            // TODO RDB: Old seqnos are never removed from the db!
+            // If the item does not exist (s.isNotFound())
+            // the seqno => key mapping could be removed; not even
+            // a tombstone remains of that item.
+            continue;
+        }
+
+        rocksdb::Slice valSlice(valueStr);
+
+        // TODO RDB: Deal with collections
+        DocKey key(reinterpret_cast<const uint8_t*>(keySlice.data() +
+                                                    sizeof(uint16_t)),
+                   keySlice.size() - sizeof(uint16_t),
+                   DocNamespace::DefaultCollection);
+
+        std::unique_ptr<Item> itm =
+                grokValSlice(ctx->vbid, key, valSlice, isMetaOnly);
+
+        if (itm->getBySeqno() > seqno) {
+            // TODO RDB: Old seqnos are never removed from the db!
+            // If the item has a newer seqno now, the stale
+            // seqno => key mapping could be removed
+            continue;
+        } else if (itm->getBySeqno() < seqno) {
+            throw std::logic_error(
+                    "RocksDBKVStore::scan: index has a higher seqno"
+                    "than the document in a snapshot!");
+        }
+
+        bool includeDeletes =
+                (ctx->docFilter == DocumentFilter::NO_DELETES) ? false : true;
+        bool onlyKeys =
+                (ctx->valFilter == ValueFilter::KEYS_ONLY) ? true : false;
+
+        if (!includeDeletes && itm->getOperation() == queue_op::del) {
+            continue;
+        }
+        int64_t byseqno = itm->getBySeqno();
+        CacheLookup lookup(key, byseqno, ctx->vbid);
+        ctx->lookup->callback(lookup);
+
+        int status = ctx->lookup->getStatus();
+
+        if (status == ENGINE_KEY_EEXISTS) {
+            ctx->lastReadSeqno = byseqno;
+            continue;
+        } else if (status == ENGINE_ENOMEM) {
+            return scan_again;
+        }
+
+        GetValue rv(std::move(itm), ENGINE_SUCCESS, -1, onlyKeys);
+        ctx->callback->callback(rv);
+        status = ctx->callback->getStatus();
+
+        if (status == ENGINE_ENOMEM) {
+            return scan_again;
+        }
+
+        ctx->lastReadSeqno = byseqno;
+    }
+
+    cb_assert(it->status().ok()); // Check for any errors found during the scan
+
+    return scan_success;
+}

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

@@ -262,10 +262,7 @@ class RocksDBKVStore : public KVStore {
                                  DocumentFilter options,
                                  ValueFilter valOptions) override;
 
-    scan_error_t scan(ScanContext* sctx) override {
-        // TODO vmx 2016-10-29: implement
-        return scan_success;
-    }
+    scan_error_t scan(ScanContext* sctx) override;
 
     void destroyScanContext(ScanContext* ctx) override {
         // TODO vmx 2016-10-29: implement

engines/ep/tests/ep_testsuite.cc

@@ -7452,14 +7452,8 @@ BaseTestCase testsuite_testcases[] = {
                  cleanup),
         TestCase("test memory condition", test_memory_condition, test_setup,
                  teardown, "max_size=2621440", prepare_ep_bucket, cleanup),
-        TestCase("warmup conf",
-                 test_warmup_conf,
-                 test_setup,
-                 teardown,
-                 NULL,
-                 /* TODO RDB: Enable after implementing RocksDBKVStore::scan */
-                 prepare_skip_broken_under_rocks,
-                 cleanup),
+        TestCase("warmup conf", test_warmup_conf, test_setup,
+                 teardown, NULL, prepare, cleanup),
         TestCase("bloomfilter conf", test_bloomfilter_conf, test_setup,
                  teardown, NULL, prepare, cleanup),
         TestCase("test bloomfilters",
@@ -7501,14 +7495,9 @@ BaseTestCase testsuite_testcases[] = {
                  cleanup),
         TestCase("test set_param message", test_set_param_message, test_setup,
                  teardown, "chk_remover_stime=1;max_size=6291456", prepare, cleanup),
-        TestCase("test warmup oom value eviction",
-                 test_warmup_oom,
-                 test_setup,
-                 teardown,
-                 "item_eviction_policy=full_eviction",
-                 /* TODO RDB: Enable after implementing RocksDBKVStore::scan */
-                 prepare_skip_broken_under_rocks,
-                 cleanup),
+        TestCase("test warmup oom value eviction", test_warmup_oom, test_setup,
+                 teardown, "item_eviction_policy=full_eviction",
+                 prepare, cleanup),
 
         // Stats tests
         TestCase("item stats", test_item_stats, test_setup, teardown, NULL,
@@ -7669,40 +7658,16 @@ BaseTestCase testsuite_testcases[] = {
                  prepare_skip_broken_under_rocks,
                  cleanup),
         // special non-Ascii keys
-        TestCase("test special char keys",
-                 test_specialKeys,
-                 test_setup,
-                 teardown,
-                 NULL,
-                 /* TODO RDB: Enable after implementing RocksDBKVStore::scan */
-                 prepare_skip_broken_under_rocks,
-                 cleanup),
-        TestCase("test binary keys",
-                 test_binKeys,
-                 test_setup,
-                 teardown,
-                 NULL,
-                 /* TODO RDB: Enable after implementing RocksDBKVStore::scan */
-                 prepare_skip_broken_under_rocks,
-                 cleanup),
+        TestCase("test special char keys", test_specialKeys, test_setup,
+                 teardown, NULL, prepare, cleanup),
+        TestCase("test binary keys", test_binKeys, test_setup,
+                 teardown, NULL, prepare, cleanup),
 
         // restart tests
-        TestCase("test restart",
-                 test_restart,
-                 test_setup,
-                 teardown,
-                 NULL,
-                 /* TODO RDB: Enable after implementing RocksDBKVStore::scan */
-                 prepare_skip_broken_under_rocks,
-                 cleanup),
-        TestCase("set+get+restart+hit (bin)",
-                 test_restart_bin_val,
-                 test_setup,
-                 teardown,
-                 NULL,
-                 /* TODO RDB: Enable after implementing RocksDBKVStore::scan */
-                 prepare_skip_broken_under_rocks,
-                 cleanup),
+        TestCase("test restart", test_restart, test_setup,
+                 teardown, NULL, prepare, cleanup),
+        TestCase("set+get+restart+hit (bin)", test_restart_bin_val,
+                 test_setup, teardown, NULL, prepare, cleanup),
         TestCase("flush+restart",
                  test_flush_restart,
                  test_setup,

engines/ep/tests/ep_testsuite_basic.cc

@@ -2293,14 +2293,8 @@ BaseTestCase testsuite_testcases[] = {
                  teardown, NULL, prepare, cleanup),
         TestCase("set/delete (invalid cas)", test_set_delete_invalid_cas,
                  test_setup, teardown, NULL, prepare, cleanup),
-        TestCase("delete/set/delete",
-                 test_delete_set,
-                 test_setup,
-                 teardown,
-                 NULL,
-                 /* TODO RDB: Enable after implementing RocksDBKVStore::scan */
-                 prepare_skip_broken_under_rocks,
-                 cleanup),
+        TestCase("delete/set/delete", test_delete_set, test_setup,
+                 teardown, NULL, prepare, cleanup),
         TestCase("get/delete with missing db file",
                  test_get_delete_missing_file,
                  test_setup,

engines/ep/tests/ep_testsuite_dcp.cc

@@ -5989,12 +5989,9 @@ BaseTestCase testsuite_testcases[] = {
         TestCase("test dcp consumer noop", test_dcp_consumer_noop, test_setup,
                  teardown, nullptr, prepare, cleanup),
         TestCase("test dcp replica stream backfill",
-                 test_dcp_replica_stream_backfill,
-                 test_setup,
-                 teardown,
+                 test_dcp_replica_stream_backfill, test_setup, teardown,
                  "chk_remover_stime=1;max_checkpoints=2",
-                 /* TODO RDB: Enable after implementing RocksDBKVStore::scan */
-                 prepare_skip_broken_under_rocks,
+                 prepare,
                  cleanup),
         TestCase("test dcp replica stream in-memory",
                  test_dcp_replica_stream_in_memory, test_setup, teardown,
@@ -6022,15 +6019,12 @@ BaseTestCase testsuite_testcases[] = {
                  prepare_skip_broken_under_rocks,
                  cleanup),
         TestCase("test producer stream request (partial)",
-                 test_dcp_producer_stream_req_partial,
-                 test_setup,
-                 teardown,
+                 test_dcp_producer_stream_req_partial, test_setup, teardown,
                  /* set chk_period to essentially infinity so it won't run
                     during this test and create extra checkpoints we don't want.*/
                  "chk_remover_stime=1;chk_max_items=100;"
                  "chk_period=1000000",
-                 /* TODO RDB: Enable after implementing RocksDBKVStore::scan */
-                 prepare_skip_broken_under_rocks,
+                 prepare,
                  cleanup),
         TestCase("test producer stream request (full merged snapshots)",
                  test_dcp_producer_stream_req_full_merged_snapshots,
@@ -6382,13 +6376,8 @@ BaseTestCase testsuite_testcases[] = {
                  test_set_dcp_param, test_setup, teardown, NULL,
                  prepare, cleanup),
         TestCase("test MB-23863 backfill deleted value",
-                 test_dcp_producer_deleted_item_backfill,
-                 test_setup,
-                 teardown,
-                 NULL,
-                 /* TODO RDB: Enable after implementing RocksDBKVStore::scan */
-                 prepare_ep_bucket_skip_broken_under_rocks,
-                 cleanup),
+                 test_dcp_producer_deleted_item_backfill, test_setup, teardown,
+                 NULL, prepare_ep_bucket, cleanup),
         TestCase("test noop mandatory",test_dcp_noop_mandatory,
                  test_setup, teardown, NULL, prepare, cleanup),