Adding upstream version 14.2.21.upstream/14.2.21upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

6 files changed, 3493 insertions, 0 deletions

diff --git a/src/test/ObjectMap/CMakeLists.txt b/src/test/ObjectMap/CMakeLists.txt
new file mode 100644
index 00000000..837ec543
--- /dev/null
+++ b/src/test/ObjectMap/CMakeLists.txt
@@ -0,0 +1,42 @@
+# ceph_test_object_map
+add_executable(ceph_test_object_map
+  test_object_map.cc
+  KeyValueDBMemory.cc
+  )
+add_ceph_unittest(ceph_test_object_map)
+target_link_libraries(ceph_test_object_map
+  os
+  ceph-common
+  ${UNITTEST_LIBS}
+  global
+  ${EXTRALIBS}
+  ${CMAKE_DL_LIBS}
+  )
+
+# ceph_test_keyvaluedb_atomicity
+add_executable(ceph_test_keyvaluedb_atomicity
+  test_keyvaluedb_atomicity.cc
+  )
+target_link_libraries(ceph_test_keyvaluedb_atomicity
+  os
+  ceph-common
+  ${UNITTEST_LIBS}
+  global
+  ${EXTRALIBS}
+  ${CMAKE_DL_LIBS}
+  )
+
+# ceph_test_keyvaluedb_iterators
+add_executable(ceph_test_keyvaluedb_iterators
+  test_keyvaluedb_iterators.cc
+  KeyValueDBMemory.cc
+  )
+target_link_libraries(ceph_test_keyvaluedb_iterators
+  os
+  ceph-common
+  ${UNITTEST_LIBS}
+  global
+  ${EXTRALIBS}
+  ${CMAKE_DL_LIBS}
+  )
+
diff --git a/src/test/ObjectMap/KeyValueDBMemory.cc b/src/test/ObjectMap/KeyValueDBMemory.cc
new file mode 100644
index 00000000..a4b2a4e1
--- /dev/null
+++ b/src/test/ObjectMap/KeyValueDBMemory.cc
@@ -0,0 +1,264 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+#include "include/encoding.h"
+#include "KeyValueDBMemory.h"
+#include <map>
+#include <set>
+#include <iostream>
+
+using namespace std;
+
+/**
+ * Iterate over the whole key space of the in-memory store
+ *
+ * @note Removing keys from the store while iterating over the store key-space
+ *	 may result in unspecified behavior.
+ *	 If one wants to safely iterate over the store while updating the
+ *	 store, one should instead use a snapshot iterator, which provides
+ *	 strong read-consistency.
+ */
+class WholeSpaceMemIterator : public KeyValueDB::WholeSpaceIteratorImpl {
+protected:
+  KeyValueDBMemory *db;
+  bool ready;
+
+  map<pair<string,string>, bufferlist>::iterator it;
+
+public:
+  explicit WholeSpaceMemIterator(KeyValueDBMemory *db) : db(db), ready(false) { }
+  ~WholeSpaceMemIterator() override { }
+
+  int seek_to_first() override {
+    if (db->db.empty()) {
+      it = db->db.end();
+      ready = false;
+      return 0;
+    }
+    it = db->db.begin();
+    ready = true;
+    return 0;
+  }
+
+  int seek_to_first(const string &prefix) override {
+    it = db->db.lower_bound(make_pair(prefix, ""));
+    if (db->db.empty() || (it == db->db.end())) {
+      it = db->db.end();
+      ready = false;
+      return 0;
+    }
+    ready = true;
+    return 0;
+  }
+
+  int seek_to_last() override {
+    it = db->db.end();
+    if (db->db.empty()) {
+      ready = false;
+      return 0;
+    }
+    --it;
+    ceph_assert(it != db->db.end());
+    ready = true;
+    return 0;
+  }
+
+  int seek_to_last(const string &prefix) override {
+    string tmp(prefix);
+    tmp.append(1, (char) 0);
+    it = db->db.upper_bound(make_pair(tmp,""));
+
+    if (db->db.empty() || (it == db->db.end())) {
+      seek_to_last();
+    }
+    else {
+      ready = true;
+      prev();
+    }
+    return 0;
+  }
+
+  int lower_bound(const string &prefix, const string &to) override {
+    it = db->db.lower_bound(make_pair(prefix,to));
+    if ((db->db.empty()) || (it == db->db.end())) {
+      it = db->db.end();
+      ready = false;
+      return 0;
+    }
+
+    ceph_assert(it != db->db.end());
+
+    ready = true;
+    return 0;
+  }
+
+  int upper_bound(const string &prefix, const string &after) override {
+    it = db->db.upper_bound(make_pair(prefix,after));
+    if ((db->db.empty()) || (it == db->db.end())) {
+      it = db->db.end();
+      ready = false;
+      return 0;
+    }
+    ceph_assert(it != db->db.end());
+    ready = true;
+    return 0;
+  }
+
+  bool valid() override {
+    return ready && (it != db->db.end());
+  }
+
+  bool begin() {
+    return ready && (it == db->db.begin());
+  }
+
+  int prev() override {
+    if (!begin() && ready)
+      --it;
+    else
+      it = db->db.end();
+    return 0;
+  }
+
+  int next() override {
+    if (valid())
+      ++it;
+    return 0;
+  }
+
+  string key() override {
+    if (valid())
+      return (*it).first.second;
+    else
+      return "";
+  }
+
+  pair<string,string> raw_key() override {
+    if (valid())
+      return (*it).first;
+    else
+      return make_pair("", "");
+  }
+  
+  bool raw_key_is_prefixed(const string &prefix) override {
+    return prefix == (*it).first.first;
+  }
+
+  bufferlist value() override {
+    if (valid())
+      return (*it).second;
+    else
+      return bufferlist();
+  }
+
+  int status() override {
+    return 0;
+  }
+};
+
+int KeyValueDBMemory::get(const string &prefix,
+			  const std::set<string> &key,
+			  map<string, bufferlist> *out) {
+  if (!exists_prefix(prefix))
+    return 0;
+
+  for (std::set<string>::const_iterator i = key.begin();
+       i != key.end();
+       ++i) {
+    pair<string,string> k(prefix, *i);
+    if (db.count(k))
+      (*out)[*i] = db[k];
+  }
+  return 0;
+}
+
+int KeyValueDBMemory::get_keys(const string &prefix,
+			       const std::set<string> &key,
+			       std::set<string> *out) {
+  if (!exists_prefix(prefix))
+    return 0;
+
+  for (std::set<string>::const_iterator i = key.begin();
+       i != key.end();
+       ++i) {
+    if (db.count(make_pair(prefix, *i)))
+      out->insert(*i);
+  }
+  return 0;
+}
+
+int KeyValueDBMemory::set(const string &prefix,
+			  const string &key,
+			  const bufferlist &bl) {
+  db[make_pair(prefix,key)] = bl;
+  return 0;
+}
+
+int KeyValueDBMemory::rmkey(const string &prefix,
+			    const string &key) {
+  db.erase(make_pair(prefix,key));
+  return 0;
+}
+
+int KeyValueDBMemory::rmkeys_by_prefix(const string &prefix) {
+  map<std::pair<string,string>,bufferlist>::iterator i;
+  i = db.lower_bound(make_pair(prefix, ""));
+  if (i == db.end())
+    return 0;
+
+  while (i != db.end()) {
+    std::pair<string,string> key = (*i).first;
+    if (key.first != prefix)
+      break;
+
+    ++i;
+    rmkey(key.first, key.second);
+  }
+  return 0;
+}
+
+int KeyValueDBMemory::rm_range_keys(const string &prefix, const string &start, const string &end) {
+  map<std::pair<string,string>,bufferlist>::iterator i;
+  i = db.lower_bound(make_pair(prefix, start));
+  if (i == db.end())
+    return 0;
+
+  while (i != db.end()) {
+    std::pair<string,string> key = (*i).first;
+    if (key.first != prefix)
+      break;
+    if (key.second >= end)
+      break;
+    ++i;
+    rmkey(key.first, key.second);
+  }
+  return 0;
+}
+
+KeyValueDB::WholeSpaceIterator KeyValueDBMemory::get_wholespace_iterator() {
+  return std::shared_ptr<KeyValueDB::WholeSpaceIteratorImpl>(
+    new WholeSpaceMemIterator(this)
+  );
+}
+
+class WholeSpaceSnapshotMemIterator : public WholeSpaceMemIterator {
+public:
+
+  /**
+   * @note
+   * We perform a copy of the db map, which is populated by bufferlists.
+   *
+   * These are designed as shallow containers, thus there is a chance that
+   * changing the underlying memory pages will lead to the iterator seeing
+   * erroneous states.
+   *
+   * Although we haven't verified this yet, there is this chance, so we should
+   * keep it in mind.
+   */
+
+  explicit WholeSpaceSnapshotMemIterator(KeyValueDBMemory *db) :
+    WholeSpaceMemIterator(db) { }
+  ~WholeSpaceSnapshotMemIterator() override {
+    delete db;
+  }
+};
+
diff --git a/src/test/ObjectMap/KeyValueDBMemory.h b/src/test/ObjectMap/KeyValueDBMemory.h
new file mode 100644
index 00000000..415399e0
--- /dev/null
+++ b/src/test/ObjectMap/KeyValueDBMemory.h
@@ -0,0 +1,190 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+#include <map>
+#include <set>
+#include <string>
+
+#include "kv/KeyValueDB.h"
+#include "include/buffer.h"
+#include "include/Context.h"
+
+using std::string;
+
+class KeyValueDBMemory : public KeyValueDB {
+public:
+  std::map<std::pair<string,string>,bufferlist> db;
+
+  KeyValueDBMemory() { }
+  explicit KeyValueDBMemory(KeyValueDBMemory *db) : db(db->db) { }
+  ~KeyValueDBMemory() override { }
+
+  int init(string _opt) override {
+    return 0;
+  }
+  int open(std::ostream &out, const vector<ColumnFamily>& cfs = {}) override {
+    return 0;
+  }
+  int create_and_open(ostream &out, const vector<ColumnFamily>& cfs = {}) override {
+    return 0;
+  }
+
+  int get(
+    const string &prefix,
+    const std::set<string> &key,
+    std::map<string, bufferlist> *out
+    ) override;
+  using KeyValueDB::get;
+
+  int get_keys(
+    const string &prefix,
+    const std::set<string> &key,
+    std::set<string> *out
+    );
+
+  int set(
+    const string &prefix,
+    const string &key,
+    const bufferlist &bl
+    );
+
+  int rmkey(
+    const string &prefix,
+    const string &key
+    );
+
+  int rmkeys_by_prefix(
+    const string &prefix
+    );
+
+  int rm_range_keys(
+      const string &prefix,
+      const string &start,
+      const string &end
+      );
+
+  class TransactionImpl_ : public TransactionImpl {
+  public:
+    list<Context *> on_commit;
+    KeyValueDBMemory *db;
+
+    explicit TransactionImpl_(KeyValueDBMemory *db) : db(db) {}
+
+
+    struct SetOp : public Context {
+      KeyValueDBMemory *db;
+      std::pair<string,string> key;
+      bufferlist value;
+      SetOp(KeyValueDBMemory *db,
+	    const std::pair<string,string> &key,
+	    const bufferlist &value)
+	: db(db), key(key), value(value) {}
+      void finish(int r) override {
+	db->set(key.first, key.second, value);
+      }
+    };
+
+    void set(const string &prefix, const string &k, const bufferlist& bl) override {
+      on_commit.push_back(new SetOp(db, std::make_pair(prefix, k), bl));
+    }
+
+    struct RmKeysOp : public Context {
+      KeyValueDBMemory *db;
+      std::pair<string,string> key;
+      RmKeysOp(KeyValueDBMemory *db,
+	       const std::pair<string,string> &key)
+	: db(db), key(key) {}
+      void finish(int r) override {
+	db->rmkey(key.first, key.second);
+      }
+    };
+
+    using KeyValueDB::TransactionImpl::rmkey;
+    using KeyValueDB::TransactionImpl::set;
+    void rmkey(const string &prefix, const string &key) override {
+      on_commit.push_back(new RmKeysOp(db, std::make_pair(prefix, key)));
+    }
+
+    struct RmKeysByPrefixOp : public Context {
+      KeyValueDBMemory *db;
+      string prefix;
+      RmKeysByPrefixOp(KeyValueDBMemory *db,
+		       const string &prefix)
+	: db(db), prefix(prefix) {}
+      void finish(int r) override {
+	db->rmkeys_by_prefix(prefix);
+      }
+    };
+    void rmkeys_by_prefix(const string &prefix) override {
+      on_commit.push_back(new RmKeysByPrefixOp(db, prefix));
+    }
+
+    struct RmRangeKeys: public Context {
+      KeyValueDBMemory *db;
+      string prefix, start, end;
+      RmRangeKeys(KeyValueDBMemory *db, const string &prefix, const string &s, const string &e)
+	: db(db), prefix(prefix), start(s), end(e) {}
+      void finish(int r) {
+	db->rm_range_keys(prefix, start, end);
+      }
+    };
+
+    void rm_range_keys(const string &prefix, const string &start, const string &end) {
+      on_commit.push_back(new RmRangeKeys(db, prefix, start, end));
+    }
+
+    int complete() {
+      for (list<Context *>::iterator i = on_commit.begin();
+	   i != on_commit.end();
+	   on_commit.erase(i++)) {
+	(*i)->complete(0);
+      }
+      return 0;
+    }
+
+    ~TransactionImpl_() override {
+      for (list<Context *>::iterator i = on_commit.begin();
+	   i != on_commit.end();
+	   on_commit.erase(i++)) {
+	delete *i;
+      }
+    }
+  };
+
+  Transaction get_transaction() override {
+    return Transaction(new TransactionImpl_(this));
+  }
+
+  int submit_transaction(Transaction trans) override {
+    return static_cast<TransactionImpl_*>(trans.get())->complete();
+  }
+
+  uint64_t get_estimated_size(map<string,uint64_t> &extras) override {
+    uint64_t total_size = 0;
+
+    for (map<pair<string,string>,bufferlist>::iterator p = db.begin();
+         p != db.end(); ++p) {
+      string prefix = p->first.first;
+      bufferlist &bl = p->second;
+
+      uint64_t sz = bl.length();
+      total_size += sz;
+      if (extras.count(prefix) == 0)
+        extras[prefix] = 0;
+      extras[prefix] += sz;
+    }
+
+    return total_size;
+  }
+
+private:
+  bool exists_prefix(const string &prefix) {
+    std::map<std::pair<string,string>,bufferlist>::iterator it;
+    it = db.lower_bound(std::make_pair(prefix, ""));
+    return ((it != db.end()) && ((*it).first.first == prefix));
+  }
+
+  friend class WholeSpaceMemIterator;
+
+public:
+  WholeSpaceIterator get_wholespace_iterator() override;
+};
diff --git a/src/test/ObjectMap/test_keyvaluedb_atomicity.cc b/src/test/ObjectMap/test_keyvaluedb_atomicity.cc
new file mode 100644
index 00000000..f93e68c4
--- /dev/null
+++ b/src/test/ObjectMap/test_keyvaluedb_atomicity.cc
@@ -0,0 +1,109 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+#include <pthread.h>
+#include "include/buffer.h"
+#include "kv/KeyValueDB.h"
+#include <sys/types.h>
+#include <dirent.h>
+#include <string>
+#include <vector>
+#include <boost/scoped_ptr.hpp>
+#include <iostream>
+#include <sstream>
+#include "stdlib.h"
+#include "global/global_context.h"
+
+using namespace std;
+
+const string CONTROL_PREFIX = "CONTROL";
+const string PRIMARY_PREFIX = "PREFIX";
+const int NUM_COPIES = 100;
+const int NUM_THREADS = 30;
+
+string prefix_gen(int i) {
+  stringstream ss;
+  ss << PRIMARY_PREFIX << "_" << i << std::endl;
+  return ss.str();
+}
+
+int verify(KeyValueDB *db) {
+  // Verify
+  {
+    map<int, KeyValueDB::Iterator> iterators;
+    for (int i = 0; i < NUM_COPIES; ++i) {
+      iterators[i] = db->get_iterator(prefix_gen(i));
+      iterators[i]->seek_to_first();
+    }
+    while (iterators.rbegin()->second->valid()) {
+      for (map<int, KeyValueDB::Iterator>::iterator i = iterators.begin();
+	   i != iterators.end();
+	   ++i) {
+	ceph_assert(i->second->valid());
+	ceph_assert(i->second->key() == iterators.rbegin()->second->key());
+	bufferlist r = i->second->value();
+	bufferlist l = iterators.rbegin()->second->value();
+	i->second->next();
+      }
+    }
+    for (map<int, KeyValueDB::Iterator>::iterator i = iterators.begin();
+	 i != iterators.end();
+	 ++i) {
+      ceph_assert(!i->second->valid());
+    }
+  }
+  return 0;
+}
+
+void *write(void *_db) {
+  KeyValueDB *db = static_cast<KeyValueDB*>(_db);
+  std::cout << "Writing..." << std::endl;
+  for (int i = 0; i < 12000; ++i) {
+    if (!(i % 10)) {
+      std::cout << "Iteration: " << i << std::endl;
+    }
+    int key_num = rand();
+    stringstream key;
+    key << key_num << std::endl;
+    map<string, bufferlist> to_set;
+    stringstream val;
+    val << i << std::endl;
+    bufferptr bp(val.str().c_str(), val.str().size() + 1);
+    to_set[key.str()].push_back(bp);
+    
+    KeyValueDB::Transaction t = db->get_transaction();
+    for (int j = 0; j < NUM_COPIES; ++j) {
+      t->set(prefix_gen(j), to_set);
+    }
+    ceph_assert(!db->submit_transaction(t));
+  }
+  return 0;
+}
+
+int main() {
+  char *path = getenv("OBJECT_MAP_PATH");
+  boost::scoped_ptr< KeyValueDB > db;
+  if (!path) {
+    std::cerr << "No path found, OBJECT_MAP_PATH undefined" << std::endl;
+    return 0;
+  }
+  string strpath(path);
+  std::cerr << "Using path: " << strpath << std::endl;
+  KeyValueDB *store = KeyValueDB::create(g_ceph_context, "leveldb", strpath);
+  ceph_assert(!store->create_and_open(std::cerr));
+  db.reset(store);
+
+  verify(db.get());
+
+  vector<pthread_t> threads(NUM_THREADS);
+  for (vector<pthread_t>::iterator i = threads.begin();
+       i != threads.end();
+       ++i) {
+    pthread_create(&*i, 0, &write, static_cast<void *>(db.get()));
+  }
+  for (vector<pthread_t>::iterator i = threads.begin();
+       i != threads.end();
+       ++i) {
+    void *tmp;
+    pthread_join(*i, &tmp);
+  }
+  verify(db.get());
+}
diff --git a/src/test/ObjectMap/test_keyvaluedb_iterators.cc b/src/test/ObjectMap/test_keyvaluedb_iterators.cc
new file mode 100644
index 00000000..15ee1f5b
--- /dev/null
+++ b/src/test/ObjectMap/test_keyvaluedb_iterators.cc
@@ -0,0 +1,1757 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+* Ceph - scalable distributed file system
+*
+* Copyright (C) 2012 Inktank, Inc.
+*
+* This is free software; you can redistribute it and/or
+* modify it under the terms of the GNU Lesser General Public
+* License version 2.1, as published by the Free Software
+* Foundation. See file COPYING.
+*/
+#include <map>
+#include <set>
+#include <deque>
+#include <boost/scoped_ptr.hpp>
+
+#include "test/ObjectMap/KeyValueDBMemory.h"
+#include "kv/KeyValueDB.h"
+#include <sys/types.h>
+#include "global/global_init.h"
+#include "common/ceph_argparse.h"
+#include "gtest/gtest.h"
+
+using namespace std;
+
+string store_path;
+
+class IteratorTest : public ::testing::Test
+{
+public:
+  boost::scoped_ptr<KeyValueDB> db;
+  boost::scoped_ptr<KeyValueDBMemory> mock;
+
+  void SetUp() override {
+    ceph_assert(!store_path.empty());
+
+    KeyValueDB *db_ptr = KeyValueDB::create(g_ceph_context, "leveldb", store_path);
+    ceph_assert(!db_ptr->create_and_open(std::cerr));
+    db.reset(db_ptr);
+    mock.reset(new KeyValueDBMemory());
+  }
+
+  void TearDown() override { }
+
+  ::testing::AssertionResult validate_db_clear(KeyValueDB *store) {
+    KeyValueDB::WholeSpaceIterator it = store->get_wholespace_iterator();
+    it->seek_to_first();
+    while (it->valid()) {
+      pair<string,string> k = it->raw_key();
+      if (mock->db.count(k)) {
+	return ::testing::AssertionFailure()
+		<< __func__
+		<< " mock store count " << mock->db.count(k)
+		<< " key(" << k.first << "," << k.second << ")";
+      }
+      it->next();
+    }
+    return ::testing::AssertionSuccess();
+  }
+
+  ::testing::AssertionResult validate_db_match() {
+    KeyValueDB::WholeSpaceIterator it = db->get_wholespace_iterator();
+    it->seek_to_first();
+    while (it->valid()) {
+      pair<string, string> k = it->raw_key();
+      if (!mock->db.count(k)) {
+	return ::testing::AssertionFailure()
+		<< __func__
+		<< " mock db.count() " << mock->db.count(k)
+		<< " key(" << k.first << "," << k.second << ")";
+      }
+
+      bufferlist it_bl = it->value();
+      bufferlist mock_bl = mock->db[k];
+
+      string it_val = _bl_to_str(it_bl);
+      string mock_val = _bl_to_str(mock_bl);
+
+      if (it_val != mock_val) {
+	return ::testing::AssertionFailure()
+		<< __func__
+		<< " key(" << k.first << "," << k.second << ")"
+		<< " mismatch db value(" << it_val << ")"
+		<< " mock value(" << mock_val << ")";
+      }
+      it->next();
+    }
+    return ::testing::AssertionSuccess();
+  }
+
+  ::testing::AssertionResult validate_iterator(
+				KeyValueDB::WholeSpaceIterator it,
+				string expected_prefix,
+				const string &expected_key,
+				const string &expected_value) {
+    if (!it->valid()) {
+      return ::testing::AssertionFailure()
+	      << __func__
+	      << " iterator not valid";
+    }
+    
+    if (!it->raw_key_is_prefixed(expected_prefix)) {
+      return ::testing::AssertionFailure()
+	      << __func__
+	      << " expected raw_key_is_prefixed() == TRUE"
+	      << " got FALSE";
+    }
+    
+    if (it->raw_key_is_prefixed("??__SomeUnexpectedValue__??")) {
+      return ::testing::AssertionFailure()
+	      << __func__
+	      << " expected raw_key_is_prefixed() == FALSE"
+	      << " got TRUE";
+    }
+ 
+    pair<string,string> key = it->raw_key();
+
+    if (expected_prefix != key.first) {
+      return ::testing::AssertionFailure()
+	      << __func__
+	      << " expected prefix '" << expected_prefix << "'"
+	      << " got prefix '" << key.first << "'";
+    }
+
+    if (expected_key != it->key()) {
+      return ::testing::AssertionFailure()
+	      << __func__
+	      << " expected key '" << expected_key << "'"
+	      << " got key '" << it->key() << "'";
+    }
+
+    if (it->key() != key.second) {
+      return ::testing::AssertionFailure()
+	      << __func__
+	      << " key '" << it->key() << "'"
+	      << " does not match"
+	      << " pair key '" << key.second << "'";
+    }
+
+    if (_bl_to_str(it->value()) != expected_value) {
+      return ::testing::AssertionFailure()
+	<< __func__
+	<< " key '(" << key.first << "," << key.second << ")''"
+	<< " expected value '" << expected_value << "'"
+	<< " got value '" << _bl_to_str(it->value()) << "'";
+    }
+
+    return ::testing::AssertionSuccess();
+  }
+
+  /**
+   * Checks if each key in the queue can be forward sequentially read from
+   * the iterator iter. All keys must be present and be prefixed with prefix,
+   * otherwise the validation will fail.
+   *
+   * Assumes that each key value must be based on the key name and generated
+   * by _gen_val().
+   */
+  void validate_prefix(KeyValueDB::WholeSpaceIterator iter,
+      string &prefix, deque<string> &keys) {
+
+    while (!keys.empty()) {
+      ASSERT_TRUE(iter->valid());
+      string expected_key = keys.front();
+      keys.pop_front();
+      string expected_value = _gen_val_str(expected_key);
+
+      ASSERT_TRUE(validate_iterator(iter, prefix,
+		  expected_key, expected_value));
+
+      iter->next();
+    }
+  }
+  /**
+   * Checks if each key in the queue can be backward sequentially read from
+   * the iterator iter. All keys must be present and be prefixed with prefix,
+   * otherwise the validation will fail.
+   *
+   * Assumes that each key value must be based on the key name and generated
+   * by _gen_val().
+   */
+  void validate_prefix_backwards(KeyValueDB::WholeSpaceIterator iter,
+      string &prefix, deque<string> &keys) {
+
+    while (!keys.empty()) {
+      ASSERT_TRUE(iter->valid());
+      string expected_key = keys.front();
+      keys.pop_front();
+      string expected_value = _gen_val_str(expected_key);
+
+      ASSERT_TRUE(validate_iterator(iter, prefix,
+		  expected_key, expected_value));
+
+      iter->prev();
+    }
+  }
+
+  void clear(KeyValueDB *store) {
+    KeyValueDB::WholeSpaceIterator it = store->get_wholespace_iterator();
+    it->seek_to_first();
+    KeyValueDB::Transaction t = store->get_transaction();
+    while (it->valid()) {
+      pair<string,string> k = it->raw_key();
+      t->rmkey(k.first, k.second);
+      it->next();
+    }
+    store->submit_transaction_sync(t);
+  }
+
+  string _bl_to_str(bufferlist val) {
+    string str(val.c_str(), val.length());
+    return str;
+  }
+
+  string _gen_val_str(const string &key) {
+    ostringstream ss;
+    ss << "##value##" << key << "##";
+    return ss.str();
+ }
+
+  bufferlist _gen_val(const string &key) {
+    bufferlist bl;
+    bl.append(_gen_val_str(key));
+    return bl;
+  }
+
+  void print_iterator(KeyValueDB::WholeSpaceIterator iter) {
+    if (!iter->valid()) {
+      std::cerr << __func__ << " iterator is not valid; stop." << std::endl;
+      return;
+    }
+
+    int i = 0;
+    while (iter->valid()) {
+      pair<string,string> k = iter->raw_key();
+      std::cerr << __func__
+		<< " pos " << (++i)
+		<< " key (" << k.first << "," << k.second << ")"
+		<< " value(" << _bl_to_str(iter->value()) << ")" << std::endl;
+      iter->next();
+    }
+  }
+
+  void print_db(KeyValueDB *store) {
+    KeyValueDB::WholeSpaceIterator it = store->get_wholespace_iterator();
+    it->seek_to_first();
+    print_iterator(it);
+  }
+};
+
+// ------- Remove Keys / Remove Keys By Prefix -------
+class RmKeysTest : public IteratorTest
+{
+public:
+  string prefix1;
+  string prefix2;
+  string prefix3;
+
+  void init(KeyValueDB *db) {
+    KeyValueDB::Transaction tx = db->get_transaction();
+
+    tx->set(prefix1, "11", _gen_val("11"));
+    tx->set(prefix1, "12", _gen_val("12"));
+    tx->set(prefix1, "13", _gen_val("13"));
+    tx->set(prefix2, "21", _gen_val("21"));
+    tx->set(prefix2, "22", _gen_val("22"));
+    tx->set(prefix2, "23", _gen_val("23"));
+    tx->set(prefix3, "31", _gen_val("31"));
+    tx->set(prefix3, "32", _gen_val("32"));
+    tx->set(prefix3, "33", _gen_val("33"));
+
+    db->submit_transaction_sync(tx);
+  }
+
+  void SetUp() override {
+    IteratorTest::SetUp();
+
+    prefix1 = "_PREFIX_1_";
+    prefix2 = "_PREFIX_2_";
+    prefix3 = "_PREFIX_3_";
+
+    clear(db.get());
+    ASSERT_TRUE(validate_db_clear(db.get()));
+    clear(mock.get());
+    ASSERT_TRUE(validate_db_match());
+
+    init(db.get());
+    init(mock.get());
+
+    ASSERT_TRUE(validate_db_match());
+  }
+
+  void TearDown() override {
+    IteratorTest::TearDown();
+  }
+
+
+  /**
+   * Test the transaction's rmkeys behavior when we remove a given prefix
+   * from the beginning of the key space, or from the end of the key space,
+   * or even simply in the middle.
+   */
+  void RmKeysByPrefix(KeyValueDB *store) {
+    // remove prefix2 ; check if prefix1 remains, and then prefix3
+    KeyValueDB::Transaction tx = store->get_transaction();
+    // remove the prefix in the middle of the key space
+    tx->rmkeys_by_prefix(prefix2);
+    store->submit_transaction_sync(tx);
+
+    deque<string> key_deque;
+    KeyValueDB::WholeSpaceIterator iter = store->get_wholespace_iterator();
+    iter->seek_to_first();
+
+    // check for prefix1
+    key_deque.push_back("11");
+    key_deque.push_back("12");
+    key_deque.push_back("13");
+    validate_prefix(iter, prefix1, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+
+    // check for prefix3
+    ASSERT_TRUE(iter->valid());
+    key_deque.clear();
+    key_deque.push_back("31");
+    key_deque.push_back("32");
+    key_deque.push_back("33");
+    validate_prefix(iter, prefix3, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+
+    ASSERT_FALSE(iter->valid());
+
+    clear(store);
+    ASSERT_TRUE(validate_db_clear(store));
+    init(store);
+
+    // remove prefix1 ; check if prefix2 and then prefix3 remain
+    tx = store->get_transaction();
+    // remove the prefix at the beginning of the key space
+    tx->rmkeys_by_prefix(prefix1);
+    store->submit_transaction_sync(tx);
+
+    iter = store->get_wholespace_iterator();
+    iter->seek_to_first();
+
+    // check for prefix2
+    key_deque.clear();
+    key_deque.push_back("21");
+    key_deque.push_back("22");
+    key_deque.push_back("23");
+    validate_prefix(iter, prefix2, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+
+    // check for prefix3
+    ASSERT_TRUE(iter->valid());
+    key_deque.clear();
+    key_deque.push_back("31");
+    key_deque.push_back("32");
+    key_deque.push_back("33");
+    validate_prefix(iter, prefix3, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+
+    ASSERT_FALSE(iter->valid());
+
+    clear(store);
+    ASSERT_TRUE(validate_db_clear(store));
+    init(store);
+
+    // remove prefix3 ; check if prefix1 and then prefix2 remain
+    tx = store->get_transaction();
+    // remove the prefix at the end of the key space
+    tx->rmkeys_by_prefix(prefix3);
+    store->submit_transaction_sync(tx);
+
+    iter = store->get_wholespace_iterator();
+    iter->seek_to_first();
+
+    // check for prefix1
+    key_deque.clear();
+    key_deque.push_back("11");
+    key_deque.push_back("12");
+    key_deque.push_back("13");
+    validate_prefix(iter, prefix1, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+
+    // check for prefix2
+    ASSERT_TRUE(iter->valid());
+    key_deque.clear();
+    key_deque.push_back("21");
+    key_deque.push_back("22");
+    key_deque.push_back("23");
+    validate_prefix(iter, prefix2, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+
+    ASSERT_FALSE(iter->valid());
+  }
+
+  /**
+   * Test how the leveldb's whole-space iterator behaves when we remove
+   * keys from the store while iterating over them.
+   */
+  void RmKeysWhileIteratingSnapshot(KeyValueDB *store,
+				    KeyValueDB::WholeSpaceIterator iter) {
+
+    SCOPED_TRACE("RmKeysWhileIteratingSnapshot");
+
+    iter->seek_to_first();
+    ASSERT_TRUE(iter->valid());
+
+    KeyValueDB::Transaction t = store->get_transaction();
+    t->rmkey(prefix1, "11");
+    t->rmkey(prefix1, "12");
+    t->rmkey(prefix2, "23");
+    t->rmkey(prefix3, "33");
+    store->submit_transaction_sync(t);
+
+    deque<string> key_deque;
+
+    // check for prefix1
+    key_deque.push_back("11");
+    key_deque.push_back("12");
+    key_deque.push_back("13");
+    validate_prefix(iter, prefix1, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+
+    // check for prefix2
+    key_deque.clear();
+    key_deque.push_back("21");
+    key_deque.push_back("22");
+    key_deque.push_back("23");
+    validate_prefix(iter, prefix2, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+
+    // check for prefix3
+    key_deque.clear();
+    key_deque.push_back("31");
+    key_deque.push_back("32");
+    key_deque.push_back("33");
+    validate_prefix(iter, prefix3, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+
+    iter->next();
+    ASSERT_FALSE(iter->valid());
+
+    // make sure those keys were removed from the store
+    KeyValueDB::WholeSpaceIterator tmp_it = store->get_wholespace_iterator();
+    tmp_it->seek_to_first();
+    ASSERT_TRUE(tmp_it->valid());
+
+    key_deque.clear();
+    key_deque.push_back("13");
+    validate_prefix(tmp_it, prefix1, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+
+    ASSERT_TRUE(tmp_it->valid());
+    key_deque.clear();
+    key_deque.push_back("21");
+    key_deque.push_back("22");
+    validate_prefix(tmp_it, prefix2, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+
+    ASSERT_TRUE(tmp_it->valid());
+    key_deque.clear();
+    key_deque.push_back("31");
+    key_deque.push_back("32");
+    validate_prefix(tmp_it, prefix3, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+
+    ASSERT_FALSE(tmp_it->valid());
+  }
+};
+
+TEST_F(RmKeysTest, RmKeysByPrefixLevelDB)
+{
+  SCOPED_TRACE("LevelDB");
+  RmKeysByPrefix(db.get());
+  ASSERT_FALSE(HasFatalFailure());
+}
+
+TEST_F(RmKeysTest, RmKeysByPrefixMockDB)
+{
+  SCOPED_TRACE("Mock DB");
+  RmKeysByPrefix(mock.get());
+  ASSERT_FALSE(HasFatalFailure());
+}
+
+/**
+ * If you refer to function RmKeysTest::RmKeysWhileIteratingSnapshot(),
+ * you will notice that we seek the iterator to the first key, and then
+ * we go on to remove several keys from the underlying store, including
+ * the first couple keys.
+ *
+ * We would expect that during this test, as soon as we removed the keys
+ * from the store, the iterator would get invalid, or cause some sort of
+ * unexpected mess.
+ *
+ * Instead, the current version of leveldb handles it perfectly, by making
+ * the iterator to use a snapshot instead of the store's real state. This
+ * way, LevelDBStore's whole-space iterator will behave much like its own
+ * whole-space snapshot iterator.
+ *
+ * However, this particular behavior of the iterator hasn't been documented
+ * on leveldb, and we should assume that it can be changed at any point in
+ * time.
+ *
+ * Therefore, we keep this test, being exactly the same as the one for the
+ * whole-space snapshot iterator, as we currently assume they should behave
+ * identically. If this test fails, at some point, and the whole-space
+ * snapshot iterator passes, then it probably means that leveldb changed
+ * how its iterator behaves.
+ */
+TEST_F(RmKeysTest, RmKeysWhileIteratingLevelDB)
+{
+  SCOPED_TRACE("LevelDB -- WholeSpaceIterator");
+  RmKeysWhileIteratingSnapshot(db.get(), db->get_wholespace_iterator());
+  ASSERT_FALSE(HasFatalFailure());
+}
+
+TEST_F(RmKeysTest, RmKeysWhileIteratingMockDB)
+{
+  std::cout << "There is no safe way to test key removal while iterating\n"
+	    << "over the mock store without using snapshots" << std::endl;
+}
+
+// ------- Set Keys / Update Values -------
+class SetKeysTest : public IteratorTest
+{
+public:
+  string prefix1;
+  string prefix2;
+
+  void init(KeyValueDB *db) {
+    KeyValueDB::Transaction tx = db->get_transaction();
+
+    tx->set(prefix1, "aaa", _gen_val("aaa"));
+    tx->set(prefix1, "ccc", _gen_val("ccc"));
+    tx->set(prefix1, "eee", _gen_val("eee"));
+    tx->set(prefix2, "vvv", _gen_val("vvv"));
+    tx->set(prefix2, "xxx", _gen_val("xxx"));
+    tx->set(prefix2, "zzz", _gen_val("zzz"));
+
+    db->submit_transaction_sync(tx);
+  }
+
+  void SetUp() override {
+    IteratorTest::SetUp();
+
+    prefix1 = "_PREFIX_1_";
+    prefix2 = "_PREFIX_2_";
+
+    clear(db.get());
+    ASSERT_TRUE(validate_db_clear(db.get()));
+    clear(mock.get());
+    ASSERT_TRUE(validate_db_match());
+
+    init(db.get());
+    init(mock.get());
+
+    ASSERT_TRUE(validate_db_match());
+  }
+
+  void TearDown() override {
+    IteratorTest::TearDown();
+  }
+
+  /**
+   * Make sure that the iterator picks on new keys added if it hasn't yet
+   * iterated away from that position.
+   * 
+   * This should only happen for the whole-space iterator when not using
+   * the snapshot version.
+   *
+   * We don't need to test the validity of all elements, but we do test
+   * inserting while moving from the first element to the last, using next()
+   * to move forward, and then we test the same behavior while iterating
+   * from the last element to the first, using prev() to move backwards.
+   */
+  void SetKeysWhileIterating(KeyValueDB *store,
+			     KeyValueDB::WholeSpaceIterator iter) {
+    iter->seek_to_first();
+    ASSERT_TRUE(iter->valid());
+    ASSERT_TRUE(validate_iterator(iter, prefix1, "aaa",
+	  	_gen_val_str("aaa")));
+    iter->next();
+    ASSERT_TRUE(iter->valid());
+    ASSERT_TRUE(validate_iterator(iter, prefix1, "ccc",
+		_bl_to_str(_gen_val("ccc"))));
+
+    // insert new key 'ddd' after 'ccc' and before 'eee'
+    KeyValueDB::Transaction tx = store->get_transaction();
+    tx->set(prefix1, "ddd", _gen_val("ddd"));
+    store->submit_transaction_sync(tx);
+
+    iter->next();
+    ASSERT_TRUE(iter->valid());
+    ASSERT_TRUE(validate_iterator(iter, prefix1, "ddd",
+		_gen_val_str("ddd")));
+
+    iter->seek_to_last();
+    ASSERT_TRUE(iter->valid());
+    tx = store->get_transaction();
+    tx->set(prefix2, "yyy", _gen_val("yyy"));
+    store->submit_transaction_sync(tx);
+
+    iter->prev();
+    ASSERT_TRUE(iter->valid());
+    ASSERT_TRUE(validate_iterator(iter, prefix2,
+				  "yyy", _gen_val_str("yyy")));
+  }
+
+  /**
+   * Make sure that the whole-space snapshot iterator does not pick on new keys
+   * added to the store since we created the iterator, thus guaranteeing
+   * read-consistency.
+   *
+   * We don't need to test the validity of all elements, but we do test
+   * inserting while moving from the first element to the last, using next()
+   * to move forward, and then we test the same behavior while iterating
+   * from the last element to the first, using prev() to move backwards.
+   */
+  void SetKeysWhileIteratingSnapshot(KeyValueDB *store,
+				     KeyValueDB::WholeSpaceIterator iter) {
+    iter->seek_to_first();
+    ASSERT_TRUE(iter->valid());
+    ASSERT_TRUE(validate_iterator(iter, prefix1, "aaa",
+	  	_gen_val_str("aaa")));
+    iter->next();
+    ASSERT_TRUE(iter->valid());
+    ASSERT_TRUE(validate_iterator(iter, prefix1, "ccc",
+		_bl_to_str(_gen_val("ccc"))));
+
+    // insert new key 'ddd' after 'ccc' and before 'eee'
+    KeyValueDB::Transaction tx = store->get_transaction();
+    tx->set(prefix1, "ddd", _gen_val("ddd"));
+    store->submit_transaction_sync(tx);
+
+    iter->next();
+    ASSERT_TRUE(iter->valid());
+    ASSERT_TRUE(validate_iterator(iter, prefix1, "eee",
+		_gen_val_str("eee")));
+
+    iter->seek_to_last();
+    ASSERT_TRUE(iter->valid());
+    tx = store->get_transaction();
+    tx->set(prefix2, "yyy", _gen_val("yyy"));
+    store->submit_transaction_sync(tx);
+
+    iter->prev();
+    ASSERT_TRUE(iter->valid());
+    ASSERT_TRUE(validate_iterator(iter, prefix2,
+				  "xxx", _gen_val_str("xxx")));
+  }
+
+  /**
+   * Make sure that the whole-space iterator is able to read values changed on
+   * the store, even after we moved to the updated position.
+   *
+   * This should only be possible when not using the whole-space snapshot
+   * version of the iterator.
+   */
+  void UpdateValuesWhileIterating(KeyValueDB *store,
+				  KeyValueDB::WholeSpaceIterator iter) {
+    iter->seek_to_first();
+    ASSERT_TRUE(iter->valid());
+    ASSERT_TRUE(validate_iterator(iter, prefix1,
+				  "aaa", _gen_val_str("aaa")));
+
+    KeyValueDB::Transaction tx = store->get_transaction();
+    tx->set(prefix1, "aaa", _gen_val("aaa_1"));
+    store->submit_transaction_sync(tx);
+
+    ASSERT_TRUE(validate_iterator(iter, prefix1,
+				  "aaa", _gen_val_str("aaa_1")));
+
+    iter->seek_to_last();
+    ASSERT_TRUE(iter->valid());
+    ASSERT_TRUE(validate_iterator(iter, prefix2,
+				  "zzz", _gen_val_str("zzz")));
+
+    tx = store->get_transaction();
+    tx->set(prefix2, "zzz", _gen_val("zzz_1"));
+    store->submit_transaction_sync(tx);
+
+    ASSERT_TRUE(validate_iterator(iter, prefix2,
+				  "zzz", _gen_val_str("zzz_1")));
+  }
+
+  /**
+   * Make sure that the whole-space iterator is able to read values changed on
+   * the store, even after we moved to the updated position.
+   *
+   * This should only be possible when not using the whole-space snapshot
+   * version of the iterator.
+   */
+  void UpdateValuesWhileIteratingSnapshot(
+				  KeyValueDB *store,
+				  KeyValueDB::WholeSpaceIterator iter) {
+    iter->seek_to_first();
+    ASSERT_TRUE(iter->valid());
+    ASSERT_TRUE(validate_iterator(iter, prefix1,
+				  "aaa", _gen_val_str("aaa")));
+
+    KeyValueDB::Transaction tx = store->get_transaction();
+    tx->set(prefix1, "aaa", _gen_val("aaa_1"));
+    store->submit_transaction_sync(tx);
+
+    ASSERT_TRUE(validate_iterator(iter, prefix1,
+				  "aaa", _gen_val_str("aaa")));
+
+    iter->seek_to_last();
+    ASSERT_TRUE(iter->valid());
+    ASSERT_TRUE(validate_iterator(iter, prefix2,
+				  "zzz", _gen_val_str("zzz")));
+
+    tx = store->get_transaction();
+    tx->set(prefix2, "zzz", _gen_val("zzz_1"));
+    store->submit_transaction_sync(tx);
+
+    ASSERT_TRUE(validate_iterator(iter, prefix2,
+				  "zzz", _gen_val_str("zzz")));
+
+    // check those values were really changed in the store
+    KeyValueDB::WholeSpaceIterator tmp_iter = store->get_wholespace_iterator();
+    tmp_iter->seek_to_first();
+    ASSERT_TRUE(tmp_iter->valid());
+    ASSERT_TRUE(validate_iterator(tmp_iter, prefix1,
+				  "aaa", _gen_val_str("aaa_1")));
+    tmp_iter->seek_to_last();
+    ASSERT_TRUE(tmp_iter->valid());
+    ASSERT_TRUE(validate_iterator(tmp_iter, prefix2,
+				  "zzz", _gen_val_str("zzz_1")));
+  }
+
+
+};
+
+TEST_F(SetKeysTest, DISABLED_SetKeysWhileIteratingLevelDB)
+{
+  SCOPED_TRACE("LevelDB: SetKeysWhileIteratingLevelDB");
+  SetKeysWhileIterating(db.get(), db->get_wholespace_iterator());
+  ASSERT_TRUE(HasFatalFailure());
+}
+
+TEST_F(SetKeysTest, SetKeysWhileIteratingMockDB)
+{
+  SCOPED_TRACE("Mock DB: SetKeysWhileIteratingMockDB");
+  SetKeysWhileIterating(mock.get(), mock->get_wholespace_iterator());
+  ASSERT_FALSE(HasFatalFailure());
+}
+
+TEST_F(SetKeysTest, DISABLED_UpdateValuesWhileIteratingLevelDB)
+{
+  SCOPED_TRACE("LevelDB: UpdateValuesWhileIteratingLevelDB");
+  UpdateValuesWhileIterating(db.get(), db->get_wholespace_iterator());
+  ASSERT_FALSE(HasFatalFailure());
+}
+
+TEST_F(SetKeysTest, UpdateValuesWhileIteratingMockDB)
+{
+  SCOPED_TRACE("MockDB: UpdateValuesWhileIteratingMockDB");
+  UpdateValuesWhileIterating(mock.get(), mock->get_wholespace_iterator());
+  ASSERT_FALSE(HasFatalFailure());
+}
+
+class BoundsTest : public IteratorTest
+{
+public:
+  string prefix1;
+  string prefix2;
+  string prefix3;
+
+  void init(KeyValueDB *store) {
+    KeyValueDB::Transaction tx = store->get_transaction();
+
+    tx->set(prefix1, "aaa", _gen_val("aaa"));
+    tx->set(prefix1, "ccc", _gen_val("ccc"));
+    tx->set(prefix1, "eee", _gen_val("eee"));
+    tx->set(prefix2, "vvv", _gen_val("vvv"));
+    tx->set(prefix2, "xxx", _gen_val("xxx"));
+    tx->set(prefix2, "zzz", _gen_val("zzz"));
+    tx->set(prefix3, "aaa", _gen_val("aaa"));
+    tx->set(prefix3, "mmm", _gen_val("mmm"));
+    tx->set(prefix3, "yyy", _gen_val("yyy"));
+
+    store->submit_transaction_sync(tx);
+  }
+
+  void SetUp() override {
+    IteratorTest::SetUp();
+
+    prefix1 = "_PREFIX_1_";
+    prefix2 = "_PREFIX_2_";
+    prefix3 = "_PREFIX_4_";
+
+    clear(db.get());
+    ASSERT_TRUE(validate_db_clear(db.get()));
+    clear(mock.get());
+    ASSERT_TRUE(validate_db_match());
+
+    init(db.get());
+    init(mock.get());
+
+    ASSERT_TRUE(validate_db_match());
+  }
+
+  void TearDown() override {
+    IteratorTest::TearDown();
+  }
+
+  void LowerBoundWithEmptyKeyOnWholeSpaceIterator(
+			    KeyValueDB::WholeSpaceIterator iter) {
+    deque<string> key_deque;
+    // see what happens when we have an empty key and try to get to the
+    // first available prefix
+    iter->lower_bound(prefix1, "");
+    ASSERT_TRUE(iter->valid());
+
+    key_deque.push_back("aaa");
+    key_deque.push_back("ccc");
+    key_deque.push_back("eee");
+    validate_prefix(iter, prefix1, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+    ASSERT_TRUE(iter->valid());
+    // if we got here without problems, then it is safe to assume the
+    // remaining prefixes are intact.
+
+    // see what happens when we have an empty key and try to get to the
+    // middle of the key-space
+    iter->lower_bound(prefix2, "");
+    ASSERT_TRUE(iter->valid());
+    key_deque.clear();
+
+    key_deque.push_back("vvv");
+    key_deque.push_back("xxx");
+    key_deque.push_back("zzz");
+    validate_prefix(iter, prefix2, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+    ASSERT_TRUE(iter->valid());
+    // if we got here without problems, then it is safe to assume the
+    // remaining prefixes are intact.
+
+    // see what happens when we have an empty key and try to get to the
+    // last prefix on the key-space
+    iter->lower_bound(prefix3, "");
+    ASSERT_TRUE(iter->valid());
+    key_deque.clear();
+
+    key_deque.push_back("aaa");
+    key_deque.push_back("mmm");
+    key_deque.push_back("yyy");
+    validate_prefix(iter, prefix3, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+    ASSERT_FALSE(iter->valid());
+    // we reached the end of the key_space, so the iterator should no longer
+    // be valid
+    
+    // see what happens when we look for an inexistent prefix, that will
+    // compare higher than the existing prefixes, with an empty key
+    // expected: reach the store's end; iterator becomes invalid
+    iter->lower_bound("_PREFIX_9_", "");
+    ASSERT_FALSE(iter->valid());
+
+    // see what happens when we look for an inexistent prefix, that will
+    // compare lower than the existing prefixes, with an empty key
+    // expected: find the first prefix; iterator is valid
+    iter->lower_bound("_PREFIX_0_", "");
+    ASSERT_TRUE(iter->valid());
+    key_deque.clear();
+    key_deque.push_back("aaa");
+    key_deque.push_back("ccc");
+    key_deque.push_back("eee");
+    validate_prefix(iter, prefix1, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+    ASSERT_TRUE(iter->valid());
+
+    // see what happens when we look for an empty prefix (that should compare
+    // lower than any existing prefixes)
+    // expected: find the first prefix; iterator is valid
+    iter->lower_bound("", "");
+    ASSERT_TRUE(iter->valid());
+    key_deque.push_back("aaa");
+    key_deque.push_back("ccc");
+    key_deque.push_back("eee");
+    validate_prefix(iter, prefix1, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+    ASSERT_TRUE(iter->valid());
+  }
+
+  void LowerBoundWithEmptyPrefixOnWholeSpaceIterator(
+			  KeyValueDB::WholeSpaceIterator iter) {
+    deque<string> key_deque;
+    // check for an empty prefix, with key 'aaa'. Since this key is shared
+    // among two different prefixes, it is relevant to check which will be
+    // found first.
+    // expected: find key (prefix1, aaa); iterator is valid
+    iter->lower_bound("", "aaa");
+    ASSERT_TRUE(iter->valid());
+
+    key_deque.push_back("aaa");
+    key_deque.push_back("ccc");
+    key_deque.push_back("eee");
+    validate_prefix(iter, prefix1, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+    ASSERT_TRUE(iter->valid());
+    // since we found prefix1, it is safe to assume that the remaining
+    // prefixes (prefix2 and prefix3) will follow
+
+    // any lower_bound operation with an empty prefix should always put the
+    // iterator in the first key in the key-space, despite what key is
+    // specified. This means that looking for ("","AAAAAAAAAA") should
+    // also position the iterator on (prefix1, aaa).
+    // expected: find key (prefix1, aaa); iterator is valid
+    iter->lower_bound("", "AAAAAAAAAA");
+    ASSERT_TRUE(iter->valid());
+    key_deque.clear();
+    key_deque.push_back("aaa");
+    key_deque.push_back("ccc");
+    key_deque.push_back("eee");
+    validate_prefix(iter, prefix1, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+    ASSERT_TRUE(iter->valid());
+
+    // note: this test is a duplicate of the one in the function above. Why?
+    // Well, because it also fits here (being its prefix empty), and one could
+    // very well run solely this test (instead of the whole battery) and would
+    // certainly expect this case to be tested.
+
+    // see what happens when we look for an empty prefix (that should compare
+    // lower than any existing prefixes)
+    // expected: find the first prefix; iterator is valid
+    iter->lower_bound("", "");
+    ASSERT_TRUE(iter->valid());
+    key_deque.push_back("aaa");
+    key_deque.push_back("ccc");
+    key_deque.push_back("eee");
+    validate_prefix(iter, prefix1, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+    ASSERT_TRUE(iter->valid());
+  }
+
+  void LowerBoundOnWholeSpaceIterator(
+			  KeyValueDB::WholeSpaceIterator iter) {
+    deque<string> key_deque;
+    // check that we find the first key in the store
+    // expected: find (prefix1, aaa); iterator is valid
+    iter->lower_bound(prefix1, "aaa");
+    ASSERT_TRUE(iter->valid());
+    key_deque.push_back("aaa");
+    validate_prefix(iter, prefix1, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+    ASSERT_TRUE(iter->valid());
+
+    // check that we find the last key in the store
+    // expected: find (prefix3, yyy); iterator is valid
+    iter->lower_bound(prefix3, "yyy");
+    ASSERT_TRUE(iter->valid());
+    key_deque.clear();
+    key_deque.push_back("yyy");
+    validate_prefix(iter, prefix3, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+    ASSERT_FALSE(iter->valid());
+
+    // check that looking for non-existent prefix '_PREFIX_0_' will
+    // always result in the first value of prefix1 (prefix1,"aaa")
+    // expected: find (prefix1, aaa); iterator is valid
+    iter->lower_bound("_PREFIX_0_", "AAAAA");
+    ASSERT_TRUE(iter->valid());
+    key_deque.clear();
+    key_deque.push_back("aaa");
+    validate_prefix(iter, prefix1, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+    ASSERT_TRUE(iter->valid());
+
+    // check that looking for non-existent prefix '_PREFIX_3_' will
+    // always result in the first value of prefix3 (prefix4,"aaa")
+    // expected: find (prefix3, aaa); iterator is valid
+    iter->lower_bound("_PREFIX_3_", "AAAAA");
+    ASSERT_TRUE(iter->valid());
+    key_deque.clear();
+    key_deque.push_back("aaa");
+    validate_prefix(iter, prefix3, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+    ASSERT_TRUE(iter->valid());
+
+    // check that looking for non-existent prefix '_PREFIX_9_' will
+    // always result in an invalid iterator.
+    // expected: iterator is invalid
+    iter->lower_bound("_PREFIX_9_", "AAAAA");
+    ASSERT_FALSE(iter->valid());
+  }
+
+  void UpperBoundWithEmptyKeyOnWholeSpaceIterator(
+			  KeyValueDB::WholeSpaceIterator iter) {
+    deque<string> key_deque;
+    // check that looking for (prefix1, "") will result in finding
+    // the first key in prefix1 (prefix1, "aaa")
+    // expected: find (prefix1, aaa); iterator is valid
+    iter->upper_bound(prefix1, "");
+    key_deque.push_back("aaa");
+    validate_prefix(iter, prefix1, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+    ASSERT_TRUE(iter->valid());
+
+    // check that looking for (prefix2, "") will result in finding
+    // the first key in prefix2 (prefix2, vvv)
+    // expected: find (prefix2, aaa); iterator is valid
+    iter->upper_bound(prefix2, "");
+    key_deque.push_back("vvv");
+    validate_prefix(iter, prefix2, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+    ASSERT_TRUE(iter->valid());
+
+
+    // check that looking for (prefix3, "") will result in finding
+    // the first key in prefix3 (prefix3, aaa)
+    // expected: find (prefix3, aaa); iterator is valid
+    iter->upper_bound(prefix3, "");
+    key_deque.push_back("aaa");
+    validate_prefix(iter, prefix3, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+    ASSERT_TRUE(iter->valid());
+
+    // see what happens when we look for an inexistent prefix, that will
+    // compare higher than the existing prefixes, with an empty key
+    // expected: reach the store's end; iterator becomes invalid
+    iter->upper_bound("_PREFIX_9_", "");
+    ASSERT_FALSE(iter->valid());
+
+    // see what happens when we look for an inexistent prefix, that will
+    // compare lower than the existing prefixes, with an empty key
+    // expected: find the first prefix; iterator is valid
+    iter->upper_bound("_PREFIX_0_", "");
+    ASSERT_TRUE(iter->valid());
+    key_deque.clear();
+    key_deque.push_back("aaa");
+    validate_prefix(iter, prefix1, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+    ASSERT_TRUE(iter->valid());
+
+    // see what happens when we look for an empty prefix (that should compare
+    // lower than any existing prefixes)
+    // expected: find the first prefix; iterator is valid
+    iter->upper_bound("", "");
+    ASSERT_TRUE(iter->valid());
+    key_deque.push_back("aaa");
+    validate_prefix(iter, prefix1, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+    ASSERT_TRUE(iter->valid());
+  }
+
+  void UpperBoundWithEmptyPrefixOnWholeSpaceIterator(
+			  KeyValueDB::WholeSpaceIterator iter) {
+    deque<string> key_deque;
+    // check for an empty prefix, with key 'aaa'. Since this key is shared
+    // among two different prefixes, it is relevant to check which will be
+    // found first.
+    // expected: find key (prefix1, aaa); iterator is valid
+    iter->upper_bound("", "aaa");
+    ASSERT_TRUE(iter->valid());
+    key_deque.push_back("aaa");
+    key_deque.push_back("ccc");
+    key_deque.push_back("eee");
+    validate_prefix(iter, prefix1, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+    ASSERT_TRUE(iter->valid());
+
+    // any upper_bound operation with an empty prefix should always put the
+    // iterator in the first key whose prefix compares greater, despite the
+    // key that is specified. This means that looking for ("","AAAAAAAAAA")
+    // should position the iterator on (prefix1, aaa).
+    // expected: find key (prefix1, aaa); iterator is valid
+    iter->upper_bound("", "AAAAAAAAAA");
+    ASSERT_TRUE(iter->valid());
+    key_deque.clear();
+    key_deque.push_back("aaa");
+    validate_prefix(iter, prefix1, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+    ASSERT_TRUE(iter->valid());
+
+    // note: this test is a duplicate of the one in the function above. Why?
+    // Well, because it also fits here (being its prefix empty), and one could
+    // very well run solely this test (instead of the whole battery) and would
+    // certainly expect this case to be tested.
+
+    // see what happens when we look for an empty prefix (that should compare
+    // lower than any existing prefixes)
+    // expected: find the first prefix; iterator is valid
+    iter->upper_bound("", "");
+    ASSERT_TRUE(iter->valid());
+    key_deque.push_back("aaa");
+    validate_prefix(iter, prefix1, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+    ASSERT_TRUE(iter->valid());
+  }
+
+  void UpperBoundOnWholeSpaceIterator(
+			  KeyValueDB::WholeSpaceIterator iter) {
+    deque<string> key_deque;
+    // check that we find the second key in the store
+    // expected: find (prefix1, ccc); iterator is valid
+    iter->upper_bound(prefix1, "bbb");
+    ASSERT_TRUE(iter->valid());
+    key_deque.push_back("ccc");
+    validate_prefix(iter, prefix1, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+    ASSERT_TRUE(iter->valid());
+
+    // check that we find the last key in the store
+    // expected: find (prefix3, yyy); iterator is valid
+    iter->upper_bound(prefix3, "xxx");
+    ASSERT_TRUE(iter->valid());
+    key_deque.clear();
+    key_deque.push_back("yyy");
+    validate_prefix(iter, prefix3, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+    ASSERT_FALSE(iter->valid());
+
+    // check that looking for non-existent prefix '_PREFIX_0_' will
+    // always result in the first value of prefix1 (prefix1,"aaa")
+    // expected: find (prefix1, aaa); iterator is valid
+    iter->upper_bound("_PREFIX_0_", "AAAAA");
+    ASSERT_TRUE(iter->valid());
+    key_deque.clear();
+    key_deque.push_back("aaa");
+    validate_prefix(iter, prefix1, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+    ASSERT_TRUE(iter->valid());
+
+    // check that looking for non-existent prefix '_PREFIX_3_' will
+    // always result in the first value of prefix3 (prefix3,"aaa")
+    // expected: find (prefix3, aaa); iterator is valid
+    iter->upper_bound("_PREFIX_3_", "AAAAA");
+    ASSERT_TRUE(iter->valid());
+    key_deque.clear();
+    key_deque.push_back("aaa");
+    validate_prefix(iter, prefix3, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+    ASSERT_TRUE(iter->valid());
+
+    // check that looking for non-existent prefix '_PREFIX_9_' will
+    // always result in an invalid iterator.
+    // expected: iterator is invalid
+    iter->upper_bound("_PREFIX_9_", "AAAAA");
+    ASSERT_FALSE(iter->valid());
+  }
+};
+
+TEST_F(BoundsTest, LowerBoundWithEmptyKeyOnWholeSpaceIteratorLevelDB)
+{
+  SCOPED_TRACE("LevelDB: Lower Bound, Empty Key, Whole-Space Iterator");
+  LowerBoundWithEmptyKeyOnWholeSpaceIterator(db->get_wholespace_iterator());
+  ASSERT_FALSE(HasFatalFailure());
+}
+
+TEST_F(BoundsTest, LowerBoundWithEmptyKeyOnWholeSpaceIteratorMockDB)
+{
+  SCOPED_TRACE("MockDB: Lower Bound, Empty Key, Whole-Space Iterator");
+  LowerBoundWithEmptyKeyOnWholeSpaceIterator(mock->get_wholespace_iterator());
+  ASSERT_FALSE(HasFatalFailure());
+}
+
+TEST_F(BoundsTest, LowerBoundWithEmptyPrefixOnWholeSpaceIteratorLevelDB)
+{
+  SCOPED_TRACE("LevelDB: Lower Bound, Empty Prefix, Whole-Space Iterator");
+  LowerBoundWithEmptyPrefixOnWholeSpaceIterator(db->get_wholespace_iterator());
+  ASSERT_FALSE(HasFatalFailure());
+}
+
+TEST_F(BoundsTest, LowerBoundWithEmptyPrefixOnWholeSpaceIteratorMockDB)
+{
+  SCOPED_TRACE("MockDB: Lower Bound, Empty Prefix, Whole-Space Iterator");
+  LowerBoundWithEmptyPrefixOnWholeSpaceIterator(mock->get_wholespace_iterator());
+  ASSERT_FALSE(HasFatalFailure());
+}
+
+TEST_F(BoundsTest, LowerBoundOnWholeSpaceIteratorLevelDB)
+{
+  SCOPED_TRACE("LevelDB: Lower Bound, Whole-Space Iterator");
+  LowerBoundOnWholeSpaceIterator(db->get_wholespace_iterator());
+  ASSERT_FALSE(HasFatalFailure());
+}
+
+TEST_F(BoundsTest, LowerBoundOnWholeSpaceIteratorMockDB)
+{
+  SCOPED_TRACE("MockDB: Lower Bound, Whole-Space Iterator");
+  LowerBoundOnWholeSpaceIterator(mock->get_wholespace_iterator());
+  ASSERT_FALSE(HasFatalFailure());
+}
+
+TEST_F(BoundsTest, UpperBoundWithEmptyKeyOnWholeSpaceIteratorLevelDB)
+{
+  SCOPED_TRACE("LevelDB: Upper Bound, Empty Key, Whole-Space Iterator");
+  UpperBoundWithEmptyKeyOnWholeSpaceIterator(db->get_wholespace_iterator());
+  ASSERT_FALSE(HasFatalFailure());
+}
+
+TEST_F(BoundsTest, UpperBoundWithEmptyKeyOnWholeSpaceIteratorMockDB)
+{
+  SCOPED_TRACE("MockDB: Upper Bound, Empty Key, Whole-Space Iterator");
+  UpperBoundWithEmptyKeyOnWholeSpaceIterator(mock->get_wholespace_iterator());
+  ASSERT_FALSE(HasFatalFailure());
+}
+
+TEST_F(BoundsTest, UpperBoundWithEmptyPrefixOnWholeSpaceIteratorLevelDB)
+{
+  SCOPED_TRACE("LevelDB: Upper Bound, Empty Prefix, Whole-Space Iterator");
+  UpperBoundWithEmptyPrefixOnWholeSpaceIterator(db->get_wholespace_iterator());
+  ASSERT_FALSE(HasFatalFailure());
+}
+
+TEST_F(BoundsTest, UpperBoundWithEmptyPrefixOnWholeSpaceIteratorMockDB)
+{
+  SCOPED_TRACE("MockDB: Upper Bound, Empty Prefix, Whole-Space Iterator");
+  UpperBoundWithEmptyPrefixOnWholeSpaceIterator(mock->get_wholespace_iterator());
+  ASSERT_FALSE(HasFatalFailure());
+}
+
+TEST_F(BoundsTest, UpperBoundOnWholeSpaceIteratorLevelDB)
+{
+  SCOPED_TRACE("LevelDB: Upper Bound, Whole-Space Iterator");
+  UpperBoundOnWholeSpaceIterator(db->get_wholespace_iterator());
+  ASSERT_FALSE(HasFatalFailure());
+}
+
+TEST_F(BoundsTest, UpperBoundOnWholeSpaceIteratorMockDB)
+{
+  SCOPED_TRACE("MockDB: Upper Bound, Whole-Space Iterator");
+  UpperBoundOnWholeSpaceIterator(mock->get_wholespace_iterator());
+  ASSERT_FALSE(HasFatalFailure());
+}
+
+
+class SeeksTest : public IteratorTest
+{
+public:
+  string prefix0;
+  string prefix1;
+  string prefix2;
+  string prefix3;
+  string prefix4;
+  string prefix5;
+
+  void init(KeyValueDB *store) {
+    KeyValueDB::Transaction tx = store->get_transaction();
+
+    tx->set(prefix1, "aaa", _gen_val("aaa"));
+    tx->set(prefix1, "ccc", _gen_val("ccc"));
+    tx->set(prefix1, "eee", _gen_val("eee"));
+    tx->set(prefix2, "vvv", _gen_val("vvv"));
+    tx->set(prefix2, "xxx", _gen_val("xxx"));
+    tx->set(prefix2, "zzz", _gen_val("zzz"));
+    tx->set(prefix4, "aaa", _gen_val("aaa"));
+    tx->set(prefix4, "mmm", _gen_val("mmm"));
+    tx->set(prefix4, "yyy", _gen_val("yyy"));
+
+    store->submit_transaction_sync(tx);
+  }
+
+  void SetUp() override {
+    IteratorTest::SetUp();
+
+    prefix0 = "_PREFIX_0_";
+    prefix1 = "_PREFIX_1_";
+    prefix2 = "_PREFIX_2_";
+    prefix3 = "_PREFIX_3_";
+    prefix4 = "_PREFIX_4_";
+    prefix5 = "_PREFIX_5_";
+
+    clear(db.get());
+    ASSERT_TRUE(validate_db_clear(db.get()));
+    clear(mock.get());
+    ASSERT_TRUE(validate_db_match());
+
+    init(db.get());
+    init(mock.get());
+
+    ASSERT_TRUE(validate_db_match());
+  }
+
+  void TearDown() override {
+    IteratorTest::TearDown();
+  }
+
+
+  void SeekToFirstOnWholeSpaceIterator(
+			  KeyValueDB::WholeSpaceIterator iter) {
+    iter->seek_to_first();
+    ASSERT_TRUE(iter->valid());
+    deque<string> key_deque;
+    key_deque.push_back("aaa");
+    key_deque.push_back("ccc");
+    key_deque.push_back("eee");
+    validate_prefix(iter, prefix1, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+    ASSERT_TRUE(iter->valid());
+  }
+
+  void SeekToFirstWithPrefixOnWholeSpaceIterator(
+			  KeyValueDB::WholeSpaceIterator iter) {
+    deque<string> key_deque;
+
+    // if the prefix is empty, we must end up seeking to the first key.
+    // expected: seek to (prefix1, aaa); iterator is valid
+    iter->seek_to_first("");
+    ASSERT_TRUE(iter->valid());
+    key_deque.push_back("aaa");
+    validate_prefix(iter, prefix1, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+    ASSERT_TRUE(iter->valid());
+
+    // try seeking to non-existent prefix that compares lower than the
+    // first available prefix
+    // expected: seek to (prefix1, aaa); iterator is valid
+    iter->seek_to_first(prefix0);
+    ASSERT_TRUE(iter->valid());
+    key_deque.clear();
+    key_deque.push_back("aaa");
+    validate_prefix(iter, prefix1, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+    ASSERT_TRUE(iter->valid());
+
+    // try seeking to non-existent prefix
+    // expected: seek to (prefix4, aaa); iterator is valid
+    iter->seek_to_first(prefix3);
+    ASSERT_TRUE(iter->valid());
+    key_deque.clear();
+    key_deque.push_back("aaa");
+    validate_prefix(iter, prefix4, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+    ASSERT_TRUE(iter->valid());
+
+    // try seeking to non-existent prefix that compares greater than the
+    // last available prefix
+    // expected: iterator is invalid
+    iter->seek_to_first(prefix5);
+    ASSERT_FALSE(iter->valid());
+
+    // try seeking to the first prefix and make sure we end up in its first
+    // position
+    // expected: seek to (prefix1,aaa); iterator is valid
+    iter->seek_to_first(prefix1);
+    ASSERT_TRUE(iter->valid());
+    key_deque.clear();
+    key_deque.push_back("aaa");
+    validate_prefix(iter, prefix1, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+    ASSERT_TRUE(iter->valid());
+
+    // try seeking to the second prefix and make sure we end up in its
+    // first position
+    // expected: seek to (prefix2,vvv); iterator is valid
+    iter->seek_to_first(prefix2);
+    ASSERT_TRUE(iter->valid());
+    key_deque.clear();
+    key_deque.push_back("vvv");
+    validate_prefix(iter, prefix2, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+    ASSERT_TRUE(iter->valid());
+
+    // try seeking to the last prefix and make sure we end up in its
+    // first position
+    // expected: seek to (prefix4,aaa); iterator is valid
+    iter->seek_to_first(prefix4);
+    ASSERT_TRUE(iter->valid());
+    key_deque.clear();
+    key_deque.push_back("aaa");
+    validate_prefix(iter, prefix4, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+    ASSERT_TRUE(iter->valid());
+  }
+
+  void SeekToLastOnWholeSpaceIterator(
+			  KeyValueDB::WholeSpaceIterator iter) {
+    deque<string> key_deque;
+    iter->seek_to_last();
+    key_deque.push_back("yyy");
+    validate_prefix(iter, prefix4, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+    ASSERT_FALSE(iter->valid());
+  }
+
+  void SeekToLastWithPrefixOnWholeSpaceIterator(
+			  KeyValueDB::WholeSpaceIterator iter) {
+    deque<string> key_deque;
+
+    // if the prefix is empty, we must end up seeking to last position
+    // that has an empty prefix, or to the previous position to the first
+    // position whose prefix compares higher than empty.
+    // expected: iterator is invalid (because (prefix1,aaa) is the first
+    //		 position that compared higher than an empty prefix)
+    iter->seek_to_last("");
+    ASSERT_FALSE(iter->valid());
+
+    // try seeking to non-existent prefix that compares lower than the
+    // first available prefix
+    // expected: iterator is invalid (because (prefix1,aaa) is the first
+    //		 position that compared higher than prefix0)
+    iter->seek_to_last(prefix0);
+    ASSERT_FALSE(iter->valid());
+
+    // try seeking to non-existent prefix
+    // expected: seek to (prefix2, zzz); iterator is valid
+    iter->seek_to_last(prefix3);
+    ASSERT_TRUE(iter->valid());
+    key_deque.clear();
+    key_deque.push_back("zzz");
+    validate_prefix(iter, prefix2, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+    ASSERT_TRUE(iter->valid());
+
+    // try seeking to non-existent prefix that compares greater than the
+    // last available prefix
+    // expected: iterator is in the last position of the store;
+    //		 i.e., (prefix4,yyy)
+    iter->seek_to_last(prefix5);
+    ASSERT_TRUE(iter->valid());
+    key_deque.clear();
+    key_deque.push_back("yyy");
+    validate_prefix(iter, prefix4, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+    ASSERT_FALSE(iter->valid());
+
+    // try seeking to the first prefix and make sure we end up in its last
+    // position
+    // expected: seek to (prefix1,eee); iterator is valid
+    iter->seek_to_last(prefix1);
+    ASSERT_TRUE(iter->valid());
+    key_deque.clear();
+    key_deque.push_back("eee");
+    validate_prefix(iter, prefix1, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+    ASSERT_TRUE(iter->valid());
+
+    // try seeking to the second prefix and make sure we end up in its
+    // last position
+    // expected: seek to (prefix2,vvv); iterator is valid
+    iter->seek_to_last(prefix2);
+    ASSERT_TRUE(iter->valid());
+    key_deque.clear();
+    key_deque.push_back("zzz");
+    validate_prefix(iter, prefix2, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+    ASSERT_TRUE(iter->valid());
+
+    // try seeking to the last prefix and make sure we end up in its
+    // last position
+    // expected: seek to (prefix4,aaa); iterator is valid
+    iter->seek_to_last(prefix4);
+    ASSERT_TRUE(iter->valid());
+    key_deque.clear();
+    key_deque.push_back("yyy");
+    validate_prefix(iter, prefix4, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+    ASSERT_FALSE(iter->valid());
+  }
+};
+
+TEST_F(SeeksTest, SeekToFirstOnWholeSpaceIteratorLevelDB) {
+  SCOPED_TRACE("LevelDB: Seek To First, Whole Space Iterator");
+  SeekToFirstOnWholeSpaceIterator(db->get_wholespace_iterator());
+  ASSERT_FALSE(HasFatalFailure());
+}
+
+TEST_F(SeeksTest, SeekToFirstOnWholeSpaceIteratorMockDB) {
+  SCOPED_TRACE("MockDB: Seek To First, Whole Space Iterator");
+  SeekToFirstOnWholeSpaceIterator(mock->get_wholespace_iterator());
+  ASSERT_FALSE(HasFatalFailure());
+}
+
+TEST_F(SeeksTest, SeekToFirstWithPrefixOnWholeSpaceIteratorLevelDB) {
+  SCOPED_TRACE("LevelDB: Seek To First, With Prefix, Whole Space Iterator");
+  SeekToFirstWithPrefixOnWholeSpaceIterator(db->get_wholespace_iterator());
+  ASSERT_FALSE(HasFatalFailure());
+}
+
+TEST_F(SeeksTest, SeekToFirstWithPrefixOnWholeSpaceIteratorMockDB) {
+  SCOPED_TRACE("MockDB: Seek To First, With Prefix, Whole Space Iterator");
+  SeekToFirstWithPrefixOnWholeSpaceIterator(mock->get_wholespace_iterator());
+  ASSERT_FALSE(HasFatalFailure());
+}
+
+TEST_F(SeeksTest, SeekToLastOnWholeSpaceIteratorLevelDB) {
+  SCOPED_TRACE("LevelDB: Seek To Last, Whole Space Iterator");
+  SeekToLastOnWholeSpaceIterator(db->get_wholespace_iterator());
+  ASSERT_FALSE(HasFatalFailure());
+}
+
+TEST_F(SeeksTest, SeekToLastOnWholeSpaceIteratorMockDB) {
+  SCOPED_TRACE("MockDB: Seek To Last, Whole Space Iterator");
+  SeekToLastOnWholeSpaceIterator(mock->get_wholespace_iterator());
+  ASSERT_FALSE(HasFatalFailure());
+}
+
+TEST_F(SeeksTest, SeekToLastWithPrefixOnWholeSpaceIteratorLevelDB) {
+  SCOPED_TRACE("LevelDB: Seek To Last, With Prefix, Whole Space Iterator");
+  SeekToLastWithPrefixOnWholeSpaceIterator(db->get_wholespace_iterator());
+  ASSERT_FALSE(HasFatalFailure());
+}
+
+TEST_F(SeeksTest, SeekToLastWithPrefixOnWholeSpaceIteratorMockDB) {
+  SCOPED_TRACE("MockDB: Seek To Last, With Prefix, Whole Space Iterator");
+  SeekToLastWithPrefixOnWholeSpaceIterator(mock->get_wholespace_iterator());
+  ASSERT_FALSE(HasFatalFailure());
+}
+
+class KeySpaceIteration : public IteratorTest
+{
+public:
+  string prefix1;
+
+  void init(KeyValueDB *store) {
+    KeyValueDB::Transaction tx = store->get_transaction();
+
+    tx->set(prefix1, "aaa", _gen_val("aaa"));
+    tx->set(prefix1, "vvv", _gen_val("vvv"));
+    tx->set(prefix1, "zzz", _gen_val("zzz"));
+
+    store->submit_transaction_sync(tx);
+  }
+
+  void SetUp() override {
+    IteratorTest::SetUp();
+
+    prefix1 = "_PREFIX_1_";
+
+    clear(db.get());
+    ASSERT_TRUE(validate_db_clear(db.get()));
+    clear(mock.get());
+    ASSERT_TRUE(validate_db_match());
+
+    init(db.get());
+    init(mock.get());
+
+    ASSERT_TRUE(validate_db_match());
+  }
+
+  void TearDown() override {
+    IteratorTest::TearDown();
+  }
+
+  void ForwardIteration(KeyValueDB::WholeSpaceIterator iter) {
+    deque<string> key_deque;
+    iter->seek_to_first();
+    key_deque.push_back("aaa");
+    key_deque.push_back("vvv");
+    key_deque.push_back("zzz");
+    validate_prefix(iter, prefix1, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+    ASSERT_FALSE(iter->valid());
+  }
+
+  void BackwardIteration(KeyValueDB::WholeSpaceIterator iter) {
+    deque<string> key_deque;
+    iter->seek_to_last();
+    key_deque.push_back("zzz");
+    key_deque.push_back("vvv");
+    key_deque.push_back("aaa");
+    validate_prefix_backwards(iter, prefix1, key_deque);
+    ASSERT_FALSE(HasFatalFailure());
+    ASSERT_FALSE(iter->valid());
+  }
+};
+
+TEST_F(KeySpaceIteration, ForwardIterationLevelDB)
+{
+  SCOPED_TRACE("LevelDB: Forward Iteration, Whole Space Iterator");
+  ForwardIteration(db->get_wholespace_iterator());
+  ASSERT_FALSE(HasFatalFailure());
+}
+
+TEST_F(KeySpaceIteration, ForwardIterationMockDB) {
+  SCOPED_TRACE("MockDB: Forward Iteration, Whole Space Iterator");
+  ForwardIteration(mock->get_wholespace_iterator());
+  ASSERT_FALSE(HasFatalFailure());
+}
+
+TEST_F(KeySpaceIteration, BackwardIterationLevelDB)
+{
+  SCOPED_TRACE("LevelDB: Backward Iteration, Whole Space Iterator");
+  BackwardIteration(db->get_wholespace_iterator());
+  ASSERT_FALSE(HasFatalFailure());
+}
+
+TEST_F(KeySpaceIteration, BackwardIterationMockDB) {
+  SCOPED_TRACE("MockDB: Backward Iteration, Whole Space Iterator");
+  BackwardIteration(mock->get_wholespace_iterator());
+  ASSERT_FALSE(HasFatalFailure());
+}
+
+class EmptyStore : public IteratorTest
+{
+public:
+  void SetUp() override {
+    IteratorTest::SetUp();
+
+    clear(db.get());
+    ASSERT_TRUE(validate_db_clear(db.get()));
+    clear(mock.get());
+    ASSERT_TRUE(validate_db_match());
+  }
+
+  void SeekToFirst(KeyValueDB::WholeSpaceIterator iter) {
+    // expected: iterator is invalid
+    iter->seek_to_first();
+    ASSERT_FALSE(iter->valid());
+  }
+
+  void SeekToFirstWithPrefix(KeyValueDB::WholeSpaceIterator iter) {
+    // expected: iterator is invalid
+    iter->seek_to_first("prefix");
+    ASSERT_FALSE(iter->valid());
+  }
+
+  void SeekToLast(KeyValueDB::WholeSpaceIterator iter) {
+    // expected: iterator is invalid
+    iter->seek_to_last();
+    ASSERT_FALSE(iter->valid());
+  }
+
+  void SeekToLastWithPrefix(KeyValueDB::WholeSpaceIterator iter) {
+    // expected: iterator is invalid
+    iter->seek_to_last("prefix");
+    ASSERT_FALSE(iter->valid());
+  }
+
+  void LowerBound(KeyValueDB::WholeSpaceIterator iter) {
+    // expected: iterator is invalid
+    iter->lower_bound("prefix", "");
+    ASSERT_FALSE(iter->valid());
+
+    // expected: iterator is invalid
+    iter->lower_bound("", "key");
+    ASSERT_FALSE(iter->valid());
+
+    // expected: iterator is invalid
+    iter->lower_bound("prefix", "key");
+    ASSERT_FALSE(iter->valid());
+  }
+
+  void UpperBound(KeyValueDB::WholeSpaceIterator iter) {
+    // expected: iterator is invalid
+    iter->upper_bound("prefix", "");
+    ASSERT_FALSE(iter->valid());
+
+    // expected: iterator is invalid
+    iter->upper_bound("", "key");
+    ASSERT_FALSE(iter->valid());
+
+    // expected: iterator is invalid
+    iter->upper_bound("prefix", "key");
+    ASSERT_FALSE(iter->valid());
+  }
+};
+
+TEST_F(EmptyStore, SeekToFirstLevelDB)
+{
+  SCOPED_TRACE("LevelDB: Empty Store, Seek To First");
+  SeekToFirst(db->get_wholespace_iterator());
+  ASSERT_FALSE(HasFatalFailure());
+}
+
+TEST_F(EmptyStore, SeekToFirstMockDB)
+{
+  SCOPED_TRACE("MockDB: Empty Store, Seek To First");
+  SeekToFirst(mock->get_wholespace_iterator());
+  ASSERT_FALSE(HasFatalFailure());
+}
+
+TEST_F(EmptyStore, SeekToFirstWithPrefixLevelDB)
+{
+  SCOPED_TRACE("LevelDB: Empty Store, Seek To First With Prefix");
+  SeekToFirstWithPrefix(db->get_wholespace_iterator());
+  ASSERT_FALSE(HasFatalFailure());
+}
+
+TEST_F(EmptyStore, SeekToFirstWithPrefixMockDB)
+{
+  SCOPED_TRACE("MockDB: Empty Store, Seek To First With Prefix");
+  SeekToFirstWithPrefix(mock->get_wholespace_iterator());
+  ASSERT_FALSE(HasFatalFailure());
+}
+
+TEST_F(EmptyStore, SeekToLastLevelDB)
+{
+  SCOPED_TRACE("LevelDB: Empty Store, Seek To Last");
+  SeekToLast(db->get_wholespace_iterator());
+  ASSERT_FALSE(HasFatalFailure());
+}
+
+TEST_F(EmptyStore, SeekToLastMockDB)
+{
+  SCOPED_TRACE("MockDB: Empty Store, Seek To Last");
+  SeekToLast(mock->get_wholespace_iterator());
+  ASSERT_FALSE(HasFatalFailure());
+}
+
+TEST_F(EmptyStore, SeekToLastWithPrefixLevelDB)
+{
+  SCOPED_TRACE("LevelDB: Empty Store, Seek To Last With Prefix");
+  SeekToLastWithPrefix(db->get_wholespace_iterator());
+  ASSERT_FALSE(HasFatalFailure());
+}
+
+TEST_F(EmptyStore, SeekToLastWithPrefixMockDB)
+{
+  SCOPED_TRACE("MockDB: Empty Store, Seek To Last With Prefix");
+  SeekToLastWithPrefix(mock->get_wholespace_iterator());
+  ASSERT_FALSE(HasFatalFailure());
+}
+
+TEST_F(EmptyStore, LowerBoundLevelDB)
+{
+  SCOPED_TRACE("LevelDB: Empty Store, Lower Bound");
+  LowerBound(db->get_wholespace_iterator());
+  ASSERT_FALSE(HasFatalFailure());
+}
+
+TEST_F(EmptyStore, LowerBoundMockDB)
+{
+  SCOPED_TRACE("MockDB: Empty Store, Lower Bound");
+  LowerBound(mock->get_wholespace_iterator());
+  ASSERT_FALSE(HasFatalFailure());
+}
+
+TEST_F(EmptyStore, UpperBoundLevelDB)
+{
+  SCOPED_TRACE("LevelDB: Empty Store, Upper Bound");
+  UpperBound(db->get_wholespace_iterator());
+  ASSERT_FALSE(HasFatalFailure());
+}
+
+TEST_F(EmptyStore, UpperBoundMockDB)
+{
+  SCOPED_TRACE("MockDB: Empty Store, Upper Bound");
+  UpperBound(mock->get_wholespace_iterator());
+  ASSERT_FALSE(HasFatalFailure());
+}
+
+
+int main(int argc, char *argv[])
+{
+  vector<const char*> args;
+  argv_to_vec(argc, (const char **) argv, args);
+
+  auto cct = global_init(NULL, args, CEPH_ENTITY_TYPE_CLIENT, CODE_ENVIRONMENT_UTILITY, CINIT_FLAG_NO_DEFAULT_CONFIG_FILE);
+  common_init_finish(g_ceph_context);
+  ::testing::InitGoogleTest(&argc, argv);
+
+  if (argc < 2) {
+    std::cerr << "Usage: " << argv[0]
+	      << "[ceph_options] [gtest_options] <store_path>" << std::endl;
+    return 1;
+  }
+  store_path = string(argv[1]);
+
+  return RUN_ALL_TESTS();
+}
diff --git a/src/test/ObjectMap/test_object_map.cc b/src/test/ObjectMap/test_object_map.cc
new file mode 100644
index 00000000..ad2c56e5
--- /dev/null
+++ b/src/test/ObjectMap/test_object_map.cc
@@ -0,0 +1,1131 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+#include <map>
+#include <set>
+#include <boost/scoped_ptr.hpp>
+
+#include "include/buffer.h"
+#include "test/ObjectMap/KeyValueDBMemory.h"
+#include "kv/KeyValueDB.h"
+#include "os/filestore/DBObjectMap.h"
+#include "os/filestore/HashIndex.h"
+#include <sys/types.h>
+#include "global/global_init.h"
+#include "common/ceph_argparse.h"
+#include <dirent.h>
+
+#include "gtest/gtest.h"
+#include "stdlib.h"
+
+using namespace std;
+
+template <typename T>
+typename T::iterator rand_choose(T &cont) {
+  if (cont.size() == 0) {
+    return cont.end();
+  }
+  int index = rand() % cont.size();
+  typename T::iterator retval = cont.begin();
+
+  for (; index > 0; --index) ++retval;
+  return retval;
+}
+
+string num_str(unsigned i) {
+  char buf[100];
+  snprintf(buf, sizeof(buf), "%.10u", i);
+  return string(buf);
+}
+
+class ObjectMapTester {
+public:
+  ObjectMap *db;
+  set<string> key_space;
+  set<string> object_name_space;
+  map<string, map<string, string> > omap;
+  map<string, string > hmap;
+  map<string, map<string, string> > xattrs;
+  unsigned seq;
+
+  ObjectMapTester() : db(0), seq(0) {}
+
+  string val_from_key(const string &object, const string &key) {
+    return object + "_" + key + "_" + num_str(seq++);
+  }
+
+  void set_key(const string &objname, const string &key, const string &value) {
+    set_key(ghobject_t(hobject_t(sobject_t(objname, CEPH_NOSNAP))),
+	    key, value);
+  }
+
+  void set_xattr(const string &objname, const string &key, const string &value) {
+    set_xattr(ghobject_t(hobject_t(sobject_t(objname, CEPH_NOSNAP))),
+	      key, value);
+  }
+
+  void set_key(ghobject_t hoid,
+	       string key, string value) {
+    map<string, bufferlist> to_write;
+    bufferptr bp(value.c_str(), value.size());
+    bufferlist bl;
+    bl.append(bp);
+    to_write.insert(make_pair(key, bl));
+    db->set_keys(hoid, to_write);
+  }
+
+  void set_keys(ghobject_t hoid, const map<string, string> &to_set) {
+    map<string, bufferlist> to_write;
+    for (auto &&i: to_set) {
+      bufferptr bp(i.second.data(), i.second.size());
+      bufferlist bl;
+      bl.append(bp);
+      to_write.insert(make_pair(i.first, bl));
+    }
+    db->set_keys(hoid, to_write);
+  }
+
+  void set_xattr(ghobject_t hoid,
+		 string key, string value) {
+    map<string, bufferlist> to_write;
+    bufferptr bp(value.c_str(), value.size());
+    bufferlist bl;
+    bl.append(bp);
+    to_write.insert(make_pair(key, bl));
+    db->set_xattrs(hoid, to_write);
+  }
+
+  void set_header(const string &objname, const string &value) {
+    set_header(ghobject_t(hobject_t(sobject_t(objname, CEPH_NOSNAP))),
+	       value);
+  }
+
+  void set_header(ghobject_t hoid,
+		  const string &value) {
+    bufferlist header;
+    header.append(bufferptr(value.c_str(), value.size() + 1));
+    db->set_header(hoid, header);
+  }
+
+  int get_header(const string &objname, string *value) {
+    return get_header(ghobject_t(hobject_t(sobject_t(objname, CEPH_NOSNAP))),
+		      value);
+  }
+
+  int get_header(ghobject_t hoid,
+		 string *value) {
+    bufferlist header;
+    int r = db->get_header(hoid, &header);
+    if (r < 0)
+      return r;
+    if (header.length())
+      *value = string(header.c_str());
+    else
+      *value = string("");
+    return 0;
+  }
+
+  int get_xattr(const string &objname, const string &key, string *value) {
+    return get_xattr(ghobject_t(hobject_t(sobject_t(objname, CEPH_NOSNAP))),
+		     key, value);
+  }
+
+  int get_xattr(ghobject_t hoid,
+		string key, string *value) {
+    set<string> to_get;
+    to_get.insert(key);
+    map<string, bufferlist> got;
+    db->get_xattrs(hoid, to_get, &got);
+    if (!got.empty()) {
+      *value = string(got.begin()->second.c_str(),
+		      got.begin()->second.length());
+      return 1;
+    } else {
+      return 0;
+    }
+  }
+
+  int get_key(const string &objname, const string &key, string *value) {
+    return get_key(ghobject_t(hobject_t(sobject_t(objname, CEPH_NOSNAP))),
+		   key, value);
+  }
+
+  int get_key(ghobject_t hoid,
+	      string key, string *value) {
+    set<string> to_get;
+    to_get.insert(key);
+    map<string, bufferlist> got;
+    db->get_values(hoid, to_get, &got);
+    if (!got.empty()) {
+      if (value) {
+	*value = string(got.begin()->second.c_str(),
+			got.begin()->second.length());
+      }
+      return 1;
+    } else {
+      return 0;
+    }
+  }
+
+  void remove_key(const string &objname, const string &key) {
+    remove_key(ghobject_t(hobject_t(sobject_t(objname, CEPH_NOSNAP))),
+	       key);
+  }
+
+  void remove_keys(const string &objname, const set<string> &to_remove) {
+    remove_keys(ghobject_t(hobject_t(sobject_t(objname, CEPH_NOSNAP))),
+                to_remove);
+  }
+
+  void remove_key(ghobject_t hoid,
+		  string key) {
+    set<string> to_remove;
+    to_remove.insert(key);
+    db->rm_keys(hoid, to_remove);
+  }
+
+  void remove_keys(ghobject_t hoid,
+                   const set<string> &to_remove) {
+    db->rm_keys(hoid, to_remove);
+  }
+
+  void remove_xattr(const string &objname, const string &key) {
+    remove_xattr(ghobject_t(hobject_t(sobject_t(objname, CEPH_NOSNAP))),
+		 key);
+  }
+
+  void remove_xattr(ghobject_t hoid,
+		    string key) {
+    set<string> to_remove;
+    to_remove.insert(key);
+    db->remove_xattrs(hoid, to_remove);
+  }
+
+  void clone(const string &objname, const string &target) {
+    clone(ghobject_t(hobject_t(sobject_t(objname, CEPH_NOSNAP))),
+	  ghobject_t(hobject_t(sobject_t(target, CEPH_NOSNAP))));
+  }
+
+  void clone(ghobject_t hoid,
+	     ghobject_t hoid2) {
+    db->clone(hoid, hoid2);
+  }
+
+  void rename(const string &objname, const string &target) {
+    rename(ghobject_t(hobject_t(sobject_t(objname, CEPH_NOSNAP))),
+	  ghobject_t(hobject_t(sobject_t(target, CEPH_NOSNAP))));
+  }
+
+  void rename(ghobject_t hoid,
+	     ghobject_t hoid2) {
+    db->rename(hoid, hoid2);
+  }
+
+  void clear(const string &objname) {
+    clear(ghobject_t(hobject_t(sobject_t(objname, CEPH_NOSNAP))));
+  }
+
+  void legacy_clone(const string &objname, const string &target) {
+    legacy_clone(ghobject_t(hobject_t(sobject_t(objname, CEPH_NOSNAP))),
+	  ghobject_t(hobject_t(sobject_t(target, CEPH_NOSNAP))));
+  }
+
+  void legacy_clone(ghobject_t hoid,
+	     ghobject_t hoid2) {
+    db->legacy_clone(hoid, hoid2);
+  }
+
+  void clear(ghobject_t hoid) {
+    db->clear(hoid);
+  }
+
+  void clear_omap(const string &objname) {
+    clear_omap(ghobject_t(hobject_t(sobject_t(objname, CEPH_NOSNAP))));
+  }
+
+  void clear_omap(const ghobject_t &objname) {
+    db->clear_keys_header(objname);
+  }
+
+  void def_init() {
+    for (unsigned i = 0; i < 10000; ++i) {
+      key_space.insert("key_" + num_str(i));
+    }
+    for (unsigned i = 0; i < 100; ++i) {
+      object_name_space.insert("name_" + num_str(i));
+    }
+  }
+
+  void init_key_set(const set<string> &keys) {
+    key_space = keys;
+  }
+
+  void init_object_name_space(const set<string> &onamespace) {
+    object_name_space = onamespace;
+  }
+
+  void auto_set_xattr(ostream &out) {
+    set<string>::iterator key = rand_choose(key_space);
+    set<string>::iterator object = rand_choose(object_name_space);
+
+    string value = val_from_key(*object, *key);
+
+    xattrs[*object][*key] = value;
+    set_xattr(*object, *key, value);
+
+    out << "auto_set_xattr " << *object << ": " << *key << " -> "
+	<< value << std::endl;
+  }
+
+  void test_set_key(const string &obj, const string &key, const string &val) {
+    omap[obj][key] = val;
+    set_key(obj, key, val);
+  }
+
+  void test_set_keys(const string &obj, const map<string, string> &to_set) {
+    for (auto &&i: to_set) {
+      omap[obj][i.first] = i.second;
+    }
+    set_keys(
+      ghobject_t(hobject_t(sobject_t(obj, CEPH_NOSNAP))),
+      to_set);
+  }
+
+  void auto_set_keys(ostream &out) {
+    set<string>::iterator object = rand_choose(object_name_space);
+
+    map<string, string> to_set;
+    unsigned amount = (rand() % 10) + 1;
+    for (unsigned i = 0; i < amount; ++i) {
+      set<string>::iterator key = rand_choose(key_space);
+      string value = val_from_key(*object, *key);
+      out << "auto_set_key " << *object << ": " << *key << " -> "
+	  << value << std::endl;
+      to_set.insert(make_pair(*key, value));
+    }
+
+
+    test_set_keys(*object, to_set);
+  }
+
+  void xattrs_on_object(const string &object, set<string> *out) {
+    if (!xattrs.count(object))
+      return;
+    const map<string, string> &xmap = xattrs.find(object)->second;
+    for (map<string, string>::const_iterator i = xmap.begin();
+	 i != xmap.end();
+	 ++i) {
+      out->insert(i->first);
+    }
+  }
+
+  void keys_on_object(const string &object, set<string> *out) {
+    if (!omap.count(object))
+      return;
+    const map<string, string> &kmap = omap.find(object)->second;
+    for (map<string, string>::const_iterator i = kmap.begin();
+	 i != kmap.end();
+	 ++i) {
+      out->insert(i->first);
+    }
+  }
+
+  void xattrs_off_object(const string &object, set<string> *out) {
+    *out = key_space;
+    set<string> xspace;
+    xattrs_on_object(object, &xspace);
+    for (set<string>::iterator i = xspace.begin();
+	 i != xspace.end();
+	 ++i) {
+      out->erase(*i);
+    }
+  }
+
+  void keys_off_object(const string &object, set<string> *out) {
+    *out = key_space;
+    set<string> kspace;
+    keys_on_object(object, &kspace);
+    for (set<string>::iterator i = kspace.begin();
+	 i != kspace.end();
+	 ++i) {
+      out->erase(*i);
+    }
+  }
+
+  int auto_check_present_xattr(ostream &out) {
+    set<string>::iterator object = rand_choose(object_name_space);
+    set<string> xspace;
+    xattrs_on_object(*object, &xspace);
+    set<string>::iterator key = rand_choose(xspace);
+    if (key == xspace.end()) {
+      return 1;
+    }
+
+    string result;
+    int r = get_xattr(*object, *key, &result);
+    if (!r) {
+      out << "auto_check_present_key: failed to find key "
+	  << *key << " on object " << *object << std::endl;
+      return 0;
+    }
+
+    if (result != xattrs[*object][*key]) {
+      out << "auto_check_present_key: for key "
+	  << *key << " on object " << *object
+	  << " found value " << result << " where we should have found "
+	  << xattrs[*object][*key] << std::endl;
+      return 0;
+    }
+
+    out << "auto_check_present_key: for key "
+	<< *key << " on object " << *object
+	<< " found value " << result << " where we should have found "
+	<< xattrs[*object][*key] << std::endl;
+    return 1;
+  }
+
+
+  int auto_check_present_key(ostream &out) {
+    set<string>::iterator object = rand_choose(object_name_space);
+    set<string> kspace;
+    keys_on_object(*object, &kspace);
+    set<string>::iterator key = rand_choose(kspace);
+    if (key == kspace.end()) {
+      return 1;
+    }
+
+    string result;
+    int r = get_key(*object, *key, &result);
+    if (!r) {
+      out << "auto_check_present_key: failed to find key "
+	  << *key << " on object " << *object << std::endl;
+      return 0;
+    }
+
+    if (result != omap[*object][*key]) {
+      out << "auto_check_present_key: for key "
+	  << *key << " on object " << *object
+	  << " found value " << result << " where we should have found "
+	  << omap[*object][*key] << std::endl;
+      return 0;
+    }
+
+    out << "auto_check_present_key: for key "
+	<< *key << " on object " << *object
+	<< " found value " << result << " where we should have found "
+	<< omap[*object][*key] << std::endl;
+    return 1;
+  }
+
+  int auto_check_absent_xattr(ostream &out) {
+    set<string>::iterator object = rand_choose(object_name_space);
+    set<string> xspace;
+    xattrs_off_object(*object, &xspace);
+    set<string>::iterator key = rand_choose(xspace);
+    if (key == xspace.end()) {
+      return 1;
+    }
+
+    string result;
+    int r = get_xattr(*object, *key, &result);
+    if (!r) {
+      out << "auto_check_absent_key: did not find key "
+	  << *key << " on object " << *object << std::endl;
+      return 1;
+    }
+
+    out << "auto_check_basent_key: for key "
+	<< *key << " on object " << *object
+	<< " found value " << result << " where we should have found nothing"
+	<< std::endl;
+    return 0;
+  }
+
+  int auto_check_absent_key(ostream &out) {
+    set<string>::iterator object = rand_choose(object_name_space);
+    set<string> kspace;
+    keys_off_object(*object, &kspace);
+    set<string>::iterator key = rand_choose(kspace);
+    if (key == kspace.end()) {
+      return 1;
+    }
+
+    string result;
+    int r = get_key(*object, *key, &result);
+    if (!r) {
+      out << "auto_check_absent_key: did not find key "
+	  << *key << " on object " << *object << std::endl;
+      return 1;
+    }
+
+    out << "auto_check_basent_key: for key "
+	<< *key << " on object " << *object
+	<< " found value " << result << " where we should have found nothing"
+	<< std::endl;
+    return 0;
+  }
+
+  void test_clone(const string &object, const string &target, ostream &out) {
+    clone(object, target);
+    if (!omap.count(object)) {
+      out << " source missing.";
+      omap.erase(target);
+    } else {
+      out << " source present.";
+      omap[target] = omap[object];
+    }
+    if (!hmap.count(object)) {
+      out << " hmap source missing." << std::endl;
+      hmap.erase(target);
+    } else {
+      out << " hmap source present." << std::endl;
+      hmap[target] = hmap[object];
+    }
+    if (!xattrs.count(object)) {
+      out << " hmap source missing." << std::endl;
+      xattrs.erase(target);
+    } else {
+      out << " hmap source present." << std::endl;
+      xattrs[target] = xattrs[object];
+    }
+  }
+
+  void auto_clone_key(ostream &out) {
+    set<string>::iterator object = rand_choose(object_name_space);
+    set<string>::iterator target = rand_choose(object_name_space);
+    while (target == object) {
+      target = rand_choose(object_name_space);
+    }
+    out << "clone " << *object << " to " << *target;
+    test_clone(*object, *target, out);
+  }
+
+  void test_remove_keys(const string &obj, const set<string> &to_remove) {
+    for (auto &&k: to_remove)
+      omap[obj].erase(k);
+    remove_keys(obj, to_remove);
+  }
+
+  void test_remove_key(const string &obj, const string &key) {
+    omap[obj].erase(key);
+    remove_key(obj, key);
+  }
+
+  void auto_remove_keys(ostream &out) {
+    set<string>::iterator object = rand_choose(object_name_space);
+    set<string> kspace;
+    keys_on_object(*object, &kspace);
+    set<string> to_remove;
+    for (unsigned i = 0; i < 3; ++i) {
+      set<string>::iterator key = rand_choose(kspace);
+      if (key == kspace.end())
+	continue;
+      out << "removing " << *key << " from " << *object << std::endl;
+      to_remove.insert(*key);
+    }
+    test_remove_keys(*object, to_remove);
+  }
+
+  void auto_remove_xattr(ostream &out) {
+    set<string>::iterator object = rand_choose(object_name_space);
+    set<string> kspace;
+    xattrs_on_object(*object, &kspace);
+    set<string>::iterator key = rand_choose(kspace);
+    if (key == kspace.end()) {
+      return;
+    }
+    out << "removing xattr " << *key << " from " << *object << std::endl;
+    xattrs[*object].erase(*key);
+    remove_xattr(*object, *key);
+  }
+
+  void auto_delete_object(ostream &out) {
+    set<string>::iterator object = rand_choose(object_name_space);
+    out << "auto_delete_object " << *object << std::endl;
+    clear(*object);
+    omap.erase(*object);
+    hmap.erase(*object);
+    xattrs.erase(*object);
+  }
+
+  void test_clear(const string &obj) {
+    clear_omap(obj);
+    omap.erase(obj);
+    hmap.erase(obj);
+  }
+
+  void auto_clear_omap(ostream &out) {
+    set<string>::iterator object = rand_choose(object_name_space);
+    out << "auto_clear_object " << *object << std::endl;
+    test_clear(*object);
+  }
+
+  void auto_write_header(ostream &out) {
+    set<string>::iterator object = rand_choose(object_name_space);
+    string header = val_from_key(*object, "HEADER");
+    out << "auto_write_header: " << *object << " -> " << header << std::endl;
+    set_header(*object, header);
+    hmap[*object] = header;
+  }
+
+  int auto_verify_header(ostream &out) {
+    set<string>::iterator object = rand_choose(object_name_space);
+    out << "verify_header: " << *object << " ";
+    string header;
+    int r = get_header(*object, &header);
+    if (r < 0) {
+      ceph_abort();
+    }
+    if (header.size() == 0) {
+      if (hmap.count(*object)) {
+	out << " failed to find header " << hmap[*object] << std::endl;
+	return 0;
+      } else {
+	out << " found no header" << std::endl;
+	return 1;
+      }
+    }
+
+    if (!hmap.count(*object)) {
+      out << " found header " << header << " should have been empty"
+	      << std::endl;
+      return 0;
+    } else if (header == hmap[*object]) {
+      out << " found correct header " << header << std::endl;
+      return 1;
+    } else {
+      out << " found incorrect header " << header
+	  << " where we should have found " << hmap[*object] << std::endl;
+      return 0;
+    }
+  }
+
+  void verify_keys(const std::string &obj, ostream &out) {
+    set<string> in_db;
+    ObjectMap::ObjectMapIterator iter = db->get_iterator(
+      ghobject_t(hobject_t(sobject_t(obj, CEPH_NOSNAP))));
+    for (iter->seek_to_first(); iter->valid(); iter->next()) {
+      in_db.insert(iter->key());
+    }
+    bool err = false;
+    for (auto &&i: omap[obj]) {
+      if (!in_db.count(i.first)) {
+	out << __func__ << ": obj " << obj << " missing key "
+	    << i.first << std::endl;
+	err = true;
+      } else {
+	in_db.erase(i.first);
+      }
+    }
+    if (!in_db.empty()) {
+      out << __func__ << ": obj " << obj << " found extra keys "
+	  << in_db << std::endl;
+      err = true;
+    }
+    ASSERT_FALSE(err);
+  }
+
+  void auto_verify_objects(ostream &out) {
+    for (auto &&i: omap) {
+      verify_keys(i.first, out);
+    }
+  }
+};
+
+class ObjectMapTest : public ::testing::Test {
+public:
+  boost::scoped_ptr< ObjectMap > db;
+  ObjectMapTester tester;
+  void SetUp() override {
+    char *path = getenv("OBJECT_MAP_PATH");
+    if (!path) {
+      db.reset(new DBObjectMap(g_ceph_context, new KeyValueDBMemory()));
+      tester.db = db.get();
+      return;
+    }
+
+    string strpath(path);
+
+    cerr << "using path " << strpath << std::endl;
+    KeyValueDB *store = KeyValueDB::create(g_ceph_context, "leveldb", strpath);
+    ceph_assert(!store->create_and_open(cerr));
+
+    db.reset(new DBObjectMap(g_ceph_context, store));
+    tester.db = db.get();
+  }
+
+  void TearDown() override {
+    std::cerr << "Checking..." << std::endl;
+    ASSERT_EQ(0, db->check(std::cerr));
+  }
+};
+
+
+int main(int argc, char **argv) {
+  vector<const char*> args;
+  argv_to_vec(argc, (const char **)argv, args);
+
+  auto cct = global_init(NULL, args, CEPH_ENTITY_TYPE_CLIENT,
+			 CODE_ENVIRONMENT_UTILITY,
+			 CINIT_FLAG_NO_DEFAULT_CONFIG_FILE);
+  common_init_finish(g_ceph_context);
+  ::testing::InitGoogleTest(&argc, argv);
+  return RUN_ALL_TESTS();
+}
+
+TEST_F(ObjectMapTest, CreateOneObject) {
+  ghobject_t hoid(hobject_t(sobject_t("foo", CEPH_NOSNAP)), 100, shard_id_t(0));
+  map<string, bufferlist> to_set;
+  string key("test");
+  string val("test_val");
+  bufferptr bp(val.c_str(), val.size());
+  bufferlist bl;
+  bl.append(bp);
+  to_set.insert(make_pair(key, bl));
+  ASSERT_EQ(db->set_keys(hoid, to_set), 0);
+
+  map<string, bufferlist> got;
+  set<string> to_get;
+  to_get.insert(key);
+  to_get.insert("not there");
+  db->get_values(hoid, to_get, &got);
+  ASSERT_EQ(got.size(), (unsigned)1);
+  ASSERT_EQ(string(got[key].c_str(), got[key].length()), val);
+
+  bufferlist header;
+  got.clear();
+  db->get(hoid, &header, &got);
+  ASSERT_EQ(got.size(), (unsigned)1);
+  ASSERT_EQ(string(got[key].c_str(), got[key].length()), val);
+  ASSERT_EQ(header.length(), (unsigned)0);
+
+  db->rm_keys(hoid, to_get);
+  got.clear();
+  db->get(hoid, &header, &got);
+  ASSERT_EQ(got.size(), (unsigned)0);
+
+  map<string, bufferlist> attrs;
+  attrs["attr1"] = bl;
+  db->set_xattrs(hoid, attrs);
+
+  db->set_header(hoid, bl);
+
+  db->clear_keys_header(hoid);
+  set<string> attrs_got;
+  db->get_all_xattrs(hoid, &attrs_got);
+  ASSERT_EQ(attrs_got.size(), 1U);
+  ASSERT_EQ(*(attrs_got.begin()), "attr1");
+  db->get(hoid, &header, &got);
+  ASSERT_EQ(got.size(), (unsigned)0);
+  ASSERT_EQ(header.length(), 0U);
+  got.clear();
+
+  db->clear(hoid);
+  db->get(hoid, &header, &got);
+  ASSERT_EQ(got.size(), (unsigned)0);
+  attrs_got.clear();
+  db->get_all_xattrs(hoid, &attrs_got);
+  ASSERT_EQ(attrs_got.size(), 0U);
+}
+
+TEST_F(ObjectMapTest, CloneOneObject) {
+  ghobject_t hoid(hobject_t(sobject_t("foo", CEPH_NOSNAP)), 200, shard_id_t(0));
+  ghobject_t hoid2(hobject_t(sobject_t("foo2", CEPH_NOSNAP)), 201, shard_id_t(1));
+
+  tester.set_key(hoid, "foo", "bar");
+  tester.set_key(hoid, "foo2", "bar2");
+  string result;
+  int r = tester.get_key(hoid, "foo", &result);
+  ASSERT_EQ(r, 1);
+  ASSERT_EQ(result, "bar");
+
+  db->clone(hoid, hoid2);
+  r = tester.get_key(hoid, "foo", &result);
+  ASSERT_EQ(r, 1);
+  ASSERT_EQ(result, "bar");
+  r = tester.get_key(hoid2, "foo", &result);
+  ASSERT_EQ(r, 1);
+  ASSERT_EQ(result, "bar");
+
+  tester.remove_key(hoid, "foo");
+  r = tester.get_key(hoid2, "foo", &result);
+  ASSERT_EQ(r, 1);
+  ASSERT_EQ(result, "bar");
+  r = tester.get_key(hoid, "foo", &result);
+  ASSERT_EQ(r, 0);
+  r = tester.get_key(hoid, "foo2", &result);
+  ASSERT_EQ(r, 1);
+  ASSERT_EQ(result, "bar2");
+
+  tester.set_key(hoid, "foo", "baz");
+  tester.remove_key(hoid, "foo");
+  r = tester.get_key(hoid, "foo", &result);
+  ASSERT_EQ(r, 0);
+
+  tester.set_key(hoid, "foo2", "baz");
+  tester.remove_key(hoid, "foo2");
+  r = tester.get_key(hoid, "foo2", &result);
+  ASSERT_EQ(r, 0);
+
+  map<string, bufferlist> got;
+  bufferlist header;
+
+  got.clear();
+  db->clear(hoid);
+  db->get(hoid, &header, &got);
+  ASSERT_EQ(got.size(), (unsigned)0);
+
+  got.clear();
+  r = db->clear(hoid2);
+  ASSERT_EQ(0, r);
+  db->get(hoid2, &header, &got);
+  ASSERT_EQ(got.size(), (unsigned)0);
+
+  tester.set_key(hoid, "baz", "bar");
+  got.clear();
+  db->get(hoid, &header, &got);
+  ASSERT_EQ(got.size(), (unsigned)1);
+  db->clear(hoid);
+  db->clear(hoid2);
+}
+
+TEST_F(ObjectMapTest, OddEvenClone) {
+  ghobject_t hoid(hobject_t(sobject_t("foo", CEPH_NOSNAP)));
+  ghobject_t hoid2(hobject_t(sobject_t("foo2", CEPH_NOSNAP)));
+
+  for (unsigned i = 0; i < 1000; ++i) {
+    tester.set_key(hoid, "foo" + num_str(i), "bar" + num_str(i));
+  }
+
+  db->clone(hoid, hoid2);
+
+  int r = 0;
+  for (unsigned i = 0; i < 1000; ++i) {
+    string result;
+    r = tester.get_key(hoid, "foo" + num_str(i), &result);
+    ASSERT_EQ(1, r);
+    ASSERT_EQ("bar" + num_str(i), result);
+    r = tester.get_key(hoid2, "foo" + num_str(i), &result);
+    ASSERT_EQ(1, r);
+    ASSERT_EQ("bar" + num_str(i), result);
+
+    if (i % 2) {
+      tester.remove_key(hoid, "foo" + num_str(i));
+    } else {
+      tester.remove_key(hoid2, "foo" + num_str(i));
+    }
+  }
+
+  for (unsigned i = 0; i < 1000; ++i) {
+    string result;
+    string result2;
+    r = tester.get_key(hoid, "foo" + num_str(i), &result);
+    int r2 = tester.get_key(hoid2, "foo" + num_str(i), &result2);
+    if (i % 2) {
+      ASSERT_EQ(0, r);
+      ASSERT_EQ(1, r2);
+      ASSERT_EQ("bar" + num_str(i), result2);
+    } else {
+      ASSERT_EQ(0, r2);
+      ASSERT_EQ(1, r);
+      ASSERT_EQ("bar" + num_str(i), result);
+    }
+  }
+
+  {
+    ObjectMap::ObjectMapIterator iter = db->get_iterator(hoid);
+    iter->seek_to_first();
+    for (unsigned i = 0; i < 1000; ++i) {
+      if (!(i % 2)) {
+	ASSERT_TRUE(iter->valid());
+	ASSERT_EQ("foo" + num_str(i), iter->key());
+	iter->next();
+      }
+    }
+  }
+
+  {
+    ObjectMap::ObjectMapIterator iter2 = db->get_iterator(hoid2);
+    iter2->seek_to_first();
+    for (unsigned i = 0; i < 1000; ++i) {
+      if (i % 2) {
+	ASSERT_TRUE(iter2->valid());
+	ASSERT_EQ("foo" + num_str(i), iter2->key());
+	iter2->next();
+      }
+    }
+  }
+
+  db->clear(hoid);
+  db->clear(hoid2);
+}
+
+TEST_F(ObjectMapTest, Rename) {
+  ghobject_t hoid(hobject_t(sobject_t("foo", CEPH_NOSNAP)));
+  ghobject_t hoid2(hobject_t(sobject_t("foo2", CEPH_NOSNAP)));
+
+  for (unsigned i = 0; i < 1000; ++i) {
+    tester.set_key(hoid, "foo" + num_str(i), "bar" + num_str(i));
+  }
+
+  db->rename(hoid, hoid2);
+  // Verify rename where target exists
+  db->clone(hoid2, hoid);
+  db->rename(hoid, hoid2);
+
+  int r = 0;
+  for (unsigned i = 0; i < 1000; ++i) {
+    string result;
+    r = tester.get_key(hoid2, "foo" + num_str(i), &result);
+    ASSERT_EQ(1, r);
+    ASSERT_EQ("bar" + num_str(i), result);
+
+    if (i % 2) {
+      tester.remove_key(hoid2, "foo" + num_str(i));
+    }
+  }
+
+  for (unsigned i = 0; i < 1000; ++i) {
+    string result;
+    r = tester.get_key(hoid2, "foo" + num_str(i), &result);
+    if (i % 2) {
+      ASSERT_EQ(0, r);
+    } else {
+      ASSERT_EQ(1, r);
+      ASSERT_EQ("bar" + num_str(i), result);
+    }
+  }
+
+  {
+    ObjectMap::ObjectMapIterator iter = db->get_iterator(hoid2);
+    iter->seek_to_first();
+    for (unsigned i = 0; i < 1000; ++i) {
+      if (!(i % 2)) {
+	ASSERT_TRUE(iter->valid());
+	ASSERT_EQ("foo" + num_str(i), iter->key());
+	iter->next();
+      }
+    }
+  }
+
+  db->clear(hoid2);
+}
+
+TEST_F(ObjectMapTest, OddEvenOldClone) {
+  ghobject_t hoid(hobject_t(sobject_t("foo", CEPH_NOSNAP)));
+  ghobject_t hoid2(hobject_t(sobject_t("foo2", CEPH_NOSNAP)));
+
+  for (unsigned i = 0; i < 1000; ++i) {
+    tester.set_key(hoid, "foo" + num_str(i), "bar" + num_str(i));
+  }
+
+  db->legacy_clone(hoid, hoid2);
+
+  int r = 0;
+  for (unsigned i = 0; i < 1000; ++i) {
+    string result;
+    r = tester.get_key(hoid, "foo" + num_str(i), &result);
+    ASSERT_EQ(1, r);
+    ASSERT_EQ("bar" + num_str(i), result);
+    r = tester.get_key(hoid2, "foo" + num_str(i), &result);
+    ASSERT_EQ(1, r);
+    ASSERT_EQ("bar" + num_str(i), result);
+
+    if (i % 2) {
+      tester.remove_key(hoid, "foo" + num_str(i));
+    } else {
+      tester.remove_key(hoid2, "foo" + num_str(i));
+    }
+  }
+
+  for (unsigned i = 0; i < 1000; ++i) {
+    string result;
+    string result2;
+    r = tester.get_key(hoid, "foo" + num_str(i), &result);
+    int r2 = tester.get_key(hoid2, "foo" + num_str(i), &result2);
+    if (i % 2) {
+      ASSERT_EQ(0, r);
+      ASSERT_EQ(1, r2);
+      ASSERT_EQ("bar" + num_str(i), result2);
+    } else {
+      ASSERT_EQ(0, r2);
+      ASSERT_EQ(1, r);
+      ASSERT_EQ("bar" + num_str(i), result);
+    }
+  }
+
+  {
+    ObjectMap::ObjectMapIterator iter = db->get_iterator(hoid);
+    iter->seek_to_first();
+    for (unsigned i = 0; i < 1000; ++i) {
+      if (!(i % 2)) {
+	ASSERT_TRUE(iter->valid());
+	ASSERT_EQ("foo" + num_str(i), iter->key());
+	iter->next();
+      }
+    }
+  }
+
+  {
+    ObjectMap::ObjectMapIterator iter2 = db->get_iterator(hoid2);
+    iter2->seek_to_first();
+    for (unsigned i = 0; i < 1000; ++i) {
+      if (i % 2) {
+	ASSERT_TRUE(iter2->valid());
+	ASSERT_EQ("foo" + num_str(i), iter2->key());
+	iter2->next();
+      }
+    }
+  }
+
+  db->clear(hoid);
+  db->clear(hoid2);
+}
+
+TEST_F(ObjectMapTest, RandomTest) {
+  tester.def_init();
+  for (unsigned i = 0; i < 5000; ++i) {
+    unsigned val = rand();
+    val <<= 8;
+    val %= 100;
+    if (!(i%100))
+      std::cout << "on op " << i
+		<< " val is " << val << std::endl;
+
+    if (val < 7) {
+      tester.auto_write_header(std::cerr);
+    } else if (val < 14) {
+      ASSERT_TRUE(tester.auto_verify_header(std::cerr));
+    } else if (val < 30) {
+      tester.auto_set_keys(std::cerr);
+    } else if (val < 42) {
+      tester.auto_set_xattr(std::cerr);
+    } else if (val < 55) {
+      ASSERT_TRUE(tester.auto_check_present_key(std::cerr));
+    } else if (val < 62) {
+      ASSERT_TRUE(tester.auto_check_present_xattr(std::cerr));
+    } else if (val < 70) {
+      ASSERT_TRUE(tester.auto_check_absent_key(std::cerr));
+    } else if (val < 72) {
+      ASSERT_TRUE(tester.auto_check_absent_xattr(std::cerr));
+    } else if (val < 73) {
+      tester.auto_clear_omap(std::cerr);
+    } else if (val < 76) {
+      tester.auto_delete_object(std::cerr);
+    } else if (val < 85) {
+      tester.auto_clone_key(std::cerr);
+    } else if (val < 92) {
+      tester.auto_remove_xattr(std::cerr);
+    } else {
+      tester.auto_remove_keys(std::cerr);
+    }
+
+    if (i % 500) {
+      tester.auto_verify_objects(std::cerr);
+    }
+  }
+}
+
+TEST_F(ObjectMapTest, RandomTestNoDeletesXattrs) {
+  tester.def_init();
+  for (unsigned i = 0; i < 5000; ++i) {
+    unsigned val = rand();
+    val <<= 8;
+    val %= 100;
+    if (!(i%100))
+      std::cout << "on op " << i
+		<< " val is " << val << std::endl;
+
+    if (val < 45) {
+      tester.auto_set_keys(std::cerr);
+    } else if (val < 90) {
+      tester.auto_remove_keys(std::cerr);
+    } else {
+      tester.auto_clone_key(std::cerr);
+    }
+
+    if (i % 500) {
+      tester.auto_verify_objects(std::cerr);
+    }
+  }
+}
+
+string num_to_key(unsigned i) {
+  char buf[100];
+  int ret = snprintf(buf, sizeof(buf), "%010u", i);
+  ceph_assert(ret > 0);
+  return string(buf, ret);
+}
+
+TEST_F(ObjectMapTest, TestMergeNewCompleteContainBug) {
+  /* This test exploits a bug in kraken and earlier where merge_new_complete
+   * could miss complete entries fully contained by a new entry.  To get this
+   * to actually result in an incorrect return value, you need to remove at
+   * least two values, one before a complete region, and one which occurs in
+   * the parent after the complete region (but within 20 not yet completed
+   * parent points of the first value).
+   */
+  for (unsigned i = 10; i < 160; i+=2) {
+    tester.test_set_key("foo", num_to_key(i), "asdf");
+  }
+  tester.test_clone("foo", "foo2", std::cout);
+  tester.test_clear("foo");
+
+  tester.test_set_key("foo2", num_to_key(15), "asdf");
+  tester.test_set_key("foo2", num_to_key(13), "asdf");
+  tester.test_set_key("foo2", num_to_key(57), "asdf");
+
+  tester.test_remove_key("foo2", num_to_key(15));
+
+  set<string> to_remove;
+  to_remove.insert(num_to_key(13));
+  to_remove.insert(num_to_key(58));
+  to_remove.insert(num_to_key(60));
+  to_remove.insert(num_to_key(62));
+  tester.test_remove_keys("foo2", to_remove);
+
+  tester.verify_keys("foo2", std::cout);
+  ASSERT_EQ(tester.get_key("foo2", num_to_key(10), nullptr), 1);
+  ASSERT_EQ(tester.get_key("foo2", num_to_key(1), nullptr), 0);
+  ASSERT_EQ(tester.get_key("foo2", num_to_key(56), nullptr), 1);
+  // this one triggers the bug
+  ASSERT_EQ(tester.get_key("foo2", num_to_key(58), nullptr), 0);
+}
+
+TEST_F(ObjectMapTest, TestIterateBug18533) {
+  /* This test starts with the one immediately above to create a pair of
+   * complete regions where one contains the other.  Then, it deletes the
+   * key at the start of the contained region.  The logic in next_parent()
+   * skips ahead to the end of the contained region, and we start copying
+   * values down again from the parent into the child -- including some
+   * that had actually been deleted.  I think this works for any removal
+   * within the outer complete region after the start of the contained
+   * region.
+   */
+  for (unsigned i = 10; i < 160; i+=2) {
+    tester.test_set_key("foo", num_to_key(i), "asdf");
+  }
+  tester.test_clone("foo", "foo2", std::cout);
+  tester.test_clear("foo");
+
+  tester.test_set_key("foo2", num_to_key(15), "asdf");
+  tester.test_set_key("foo2", num_to_key(13), "asdf");
+  tester.test_set_key("foo2", num_to_key(57), "asdf");
+  tester.test_set_key("foo2", num_to_key(91), "asdf");
+
+  tester.test_remove_key("foo2", num_to_key(15));
+
+  set<string> to_remove;
+  to_remove.insert(num_to_key(13));
+  to_remove.insert(num_to_key(58));
+  to_remove.insert(num_to_key(60));
+  to_remove.insert(num_to_key(62));
+  to_remove.insert(num_to_key(82));
+  to_remove.insert(num_to_key(84));
+  tester.test_remove_keys("foo2", to_remove);
+
+  //tester.test_remove_key("foo2", num_to_key(15)); also does the trick
+  tester.test_remove_key("foo2", num_to_key(80));
+
+  // the iterator in verify_keys will return an extra value
+  tester.verify_keys("foo2", std::cout);
+}
+